services

Documentation

Index

• Variables
• func Consume(ctx context.Context, group sarama.ConsumerGroup, settings *config.Settings, ...) error
• func GetWeekNum(t time.Time) int
• func GetWeekNumForCron(t time.Time) int
• func NumToWeekEnd(n int) time.Time
• func NumToWeekStart(n int) time.Time
• type ActivityResp
• type Client
  • func (c *Client) BatchTransfer(requestID string, users []common.Address, values []*big.Int, ...) error
  • func (c *Client) TransferUserTokens(ctx context.Context, week int) error
• type ContractEventStreamConsumer
  • func NewEventConsumer(db db.Store, logger *zerolog.Logger, tc *TokenConfig) (*ContractEventStreamConsumer, error)
  • func (c *ContractEventStreamConsumer) Cleanup(sarama.ConsumerGroupSession) error
  • func (c *ContractEventStreamConsumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (c *ContractEventStreamConsumer) Setup(sarama.ConsumerGroupSession) error
• type DeviceActivityClient
• type DeviceData
• type DeviceDataClient
  • func NewDeviceDataClient(settings *config.Settings) DeviceDataClient
• type DeviceIntegrationsResp
• type DeviceListResp
• type DevicesClient
• type IntegrationsGetter
• type LevelInfo
  • func GetLevel(connStreak int) *LevelInfo
• type RewardsTask
  • func (t *RewardsTask) Calculate(issuanceWeek int) error
• type StreakInput
• type StreakOutput
  • func ComputeStreak(i StreakInput) StreakOutput
  • func FakeStreak(connectionStreak int) StreakOutput
• type TokenConfig
• type Transfer
  • func NewTokenTransferService(settings *config.Settings, producer sarama.SyncProducer, ...) Transfer
• type TransferStatusProcessor
  • func NewStatusProcessor(pdb db.Store, logger *zerolog.Logger) (*TransferStatusProcessor, error)

Variables

var LevelInfos = []*LevelInfo{
	{Level: 1, MinWeeks: 0, Points: 0},
	{Level: 2, MinWeeks: 4, Points: 1000},
	{Level: 3, MinWeeks: 21, Points: 2000},
	{Level: 4, MinWeeks: 36, Points: 3000},
}

Functions

func Consume

func Consume(ctx context.Context, group sarama.ConsumerGroup, settings *config.Settings, statusProc *TransferStatusProcessor) error

func GetWeekNum

func GetWeekNum(t time.Time) int

GetWeekNum calculates the number of the week in which the given time lies for DIMO point issuance, which at the time of writing starts at 2022-01-31 05:00 UTC. Indexing is zero-based.

func GetWeekNumForCron

func GetWeekNumForCron(t time.Time) int

GetWeekNumForCron calculates the week number for the current run of the cron job. We expect the job to run every Monday at 05:00 UTC, but due to skew we just round the time.

func NumToWeekEnd

func NumToWeekEnd(n int) time.Time

func NumToWeekStart

func NumToWeekStart(n int) time.Time

Types

type ActivityResp

type ActivityResp struct {
	Hits struct {
		Hits []struct {
			Source struct {
				Data struct {
					Timestamp time.Time `json:"timestamp"`
				} `json:"data"`
			} `json:"_source"`
		} `json:"hits"`
	} `json:"hits"`
}

type Client

type Client struct {
	Producer     sarama.SyncProducer
	Consumer     sarama.ConsumerGroup
	RequestTopic string
	StatusTopic  string

	ContractAddress common.Address
	// contains filtered or unexported fields
}

func (*Client) BatchTransfer

func (c *Client) BatchTransfer(requestID string, users []common.Address, values []*big.Int, vehicleIds []*big.Int) error

func (*Client) TransferUserTokens

func (c *Client) TransferUserTokens(ctx context.Context, week int) error

type ContractEventStreamConsumer

type ContractEventStreamConsumer struct {
	Db db.Store
	// contains filtered or unexported fields
}

func NewEventConsumer

func NewEventConsumer(db db.Store, logger *zerolog.Logger, tc *TokenConfig) (*ContractEventStreamConsumer, error)

func (*ContractEventStreamConsumer) Cleanup

func (c *ContractEventStreamConsumer) Cleanup(sarama.ConsumerGroupSession) error

func (*ContractEventStreamConsumer) ConsumeClaim

func (c *ContractEventStreamConsumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

func (*ContractEventStreamConsumer) Setup

func (c *ContractEventStreamConsumer) Setup(sarama.ConsumerGroupSession) error

type DeviceActivityClient

type DeviceActivityClient interface {
	DescribeActiveDevices(start, end time.Time) ([]*DeviceData, error)
}

type DeviceData

type DeviceData struct {
	ID           string
	Integrations []string
	Serials      []string
}

type DeviceDataClient

type DeviceDataClient interface {
	GetLastActivity(userDeviceID string) (lastActivity time.Time, seen bool, err error)
	DescribeActiveDevices(start, end time.Time) ([]*DeviceData, error)
	GetIntegrations(userDeviceID string, start, end time.Time) (ints []string, err error)
}

func NewDeviceDataClient

func NewDeviceDataClient(settings *config.Settings) DeviceDataClient

type DeviceIntegrationsResp

type DeviceIntegrationsResp struct {
	Aggregations struct {
		Integrations struct {
			Buckets []struct {
				Key string `json:"key"`
			} `json:"buckets"`
		} `json:"integrations"`
		UnitIDs struct {
			Buckets []struct {
				Key string `json:"key"`
			} `json:"buckets"`
		} `json:"unit_ids"`
	} `json:"aggregations"`
}

type DeviceListResp

type DeviceListResp struct {
	Aggregations struct {
		ActiveDevices struct {
			AfterKey struct {
				Subject *string `json:"subject"`
			} `json:"after_key"`
			Buckets []struct {
				Key struct {
					Subject string `json:"subject"`
				} `json:"key"`
				Integrations struct {
					Buckets []struct {
						Key string `json:"key"`
					} `json:"buckets"`
				} `json:"integrations"`
				UnitIDs struct {
					Buckets []struct {
						Key string `json:"key"`
					} `json:"buckets"`
				} `json:"unit_ids"`
			} `json:"buckets"`
		} `json:"active_devices"`
	} `json:"aggregations"`
}

type DevicesClient

type DevicesClient interface {
	GetUserDevice(ctx context.Context, in *pb_devices.GetUserDeviceRequest, opts ...grpc.CallOption) (*pb_devices.UserDevice, error)
}

type IntegrationsGetter

type IntegrationsGetter interface {
	GetIntegrations(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*pb_defs.GetIntegrationResponse, error)
}

type LevelInfo

type LevelInfo struct {
	Level    int
	MinWeeks int
	Points   int
}

func GetLevel

func GetLevel(connStreak int) *LevelInfo

type RewardsTask

type RewardsTask struct {
	Logger          *zerolog.Logger
	DataService     DeviceActivityClient
	DB              db.Store
	TransferService Transfer
	DevicesClient   DevicesClient
	DefsClient      IntegrationsGetter
}

func (*RewardsTask) Calculate

func (t *RewardsTask) Calculate(issuanceWeek int) error

type StreakInput

type StreakInput struct {
	ConnectedThisWeek           bool
	ExistingConnectionStreak    int
	ExistingDisconnectionStreak int
}

type StreakOutput

type StreakOutput struct {
	DisconnectionStreak int
	ConnectionStreak    int
	Points              int
}

func ComputeStreak

func ComputeStreak(i StreakInput) StreakOutput

func FakeStreak

func FakeStreak(connectionStreak int) StreakOutput

type TokenConfig

type TokenConfig struct {
	Tokens []struct {
		ChainID int64          `yaml:"chainId"`
		Address common.Address `yaml:"address"`
		RPCURL  string         `yaml:"rpcUrl"`
	} `yaml:"tokens"`
}

type Transfer

type Transfer interface {
	// BatchTransfer(requestID string, users []common.Address, values []*big.Int, vehicleIds []string) error
	TransferUserTokens(ctx context.Context, week int) error
}

func NewTokenTransferService

func NewTokenTransferService(
	settings *config.Settings,
	producer sarama.SyncProducer,
	contractAddress common.Address,

	db db.Store) Transfer

type TransferStatusProcessor

type TransferStatusProcessor struct {
	ABI                    *abi.ABI
	DB                     db.Store
	Logger                 *zerolog.Logger
	DidntQualifyEvent      abi.Event
	TokensTransferredEvent abi.Event
}

func NewStatusProcessor

func NewStatusProcessor(pdb db.Store, logger *zerolog.Logger) (*TransferStatusProcessor, error)