app

Documentation

Index

• Variables
• type Account
  • func NewAccount(address string) (*Account, error)
  • func (a *Account) Hex() string
• type CacheDuration
• type Column
• type CreateVaultParams
• type DBManager
  • func NewDBManager(dbDir string, schemas []TableSchema, windowInterval time.Duration, ...) *DBManager
  • func (dbm *DBManager) Close()
  • func (dbm *DBManager) Export(ctx context.Context, exportPath string) ([]string, error)
  • func (dbm *DBManager) NewDB(ctx context.Context) error
  • func (dbm *DBManager) Replay(ctx context.Context, tx *pgrepl.Tx) error
  • func (dbm *DBManager) UploadAll(ctx context.Context) error
  • func (dbm *DBManager) UploadAt(ctx context.Context, exportPath string) error
• type EventInfo
• type ListVaultEventsParams
• type ListVaultsParams
• type Replicator
• type RetrieveEventParams
• type Retriever
  • func NewRetriever(provider VaultsProvider, timeout int64) *Retriever
  • func (r *Retriever) Retrieve(ctx context.Context, c cid.Cid, output string) error
• type TableSchema
• type Timestamp
  • func NewTimestamp(t time.Time) Timestamp
  • func ParseTimestamp(ts string) (Timestamp, error)
  • func (t *Timestamp) Seconds() int64
• type Vault
• type VaultsProvider
• type VaultsStreamer
  • func NewVaultsStreamer(ns string, r Replicator, dbm *DBManager) *VaultsStreamer
  • func (b *VaultsStreamer) Run(ctx context.Context) error
• type VaultsUploader
  • func NewVaultsUploader(ns string, rel string, bp VaultsProvider, pk *ecdsa.PrivateKey) *VaultsUploader
  • func (bu *VaultsUploader) Upload(ctx context.Context, filepath string, progress io.Writer, ts Timestamp, ...) error
• type WriteVaultEventParams

Variables

var ErrNotFoundInCache = errors.New("not found in cache")

ErrNotFoundInCache is an error when file is not found in cache.

Types

type Account

type Account struct {
	// contains filtered or unexported fields
}

Account represents an account.

func NewAccount

func NewAccount(address string) (*Account, error)

NewAccount creates a new account.

func (*Account) Hex

func (a *Account) Hex() string

Hex returns the hex-enconded address.

type CacheDuration

type CacheDuration uint32

CacheDuration how long data stays in cache in minutes.

type Column

type Column struct {
	Name, Typ         string
	IsNull, IsPrimary bool
}

Column represents a column in a table being replicated.

type CreateVaultParams

type CreateVaultParams struct {
	Vault         Vault
	Account       *Account
	CacheDuration CacheDuration
}

CreateVaultParams ...

type DBManager

type DBManager struct {
	// contains filtered or unexported fields
}

DBManager manages a duckdb database.

func NewDBManager

func NewDBManager(
	dbDir string, schemas []TableSchema, windowInterval time.Duration, uploader *VaultsUploader,
) *DBManager

NewDBManager creates a new DBManager.

func (*DBManager) Close

func (dbm *DBManager) Close()

Close closes the current db.

func (*DBManager) Export

func (dbm *DBManager) Export(ctx context.Context, exportPath string) ([]string, error)

Export exports the current db to a parquet file at the given path.

func (*DBManager) NewDB

func (dbm *DBManager) NewDB(ctx context.Context) error

NewDB creates a new duckdb database at the <ts>.db path.

func (*DBManager) Replay

func (dbm *DBManager) Replay(ctx context.Context, tx *pgrepl.Tx) error

Replay replays a WAL record onto the current db. If the window has passed, it replaces the current db with a new one. The current db is exported and uploaded before new db is ready to be used.

func (*DBManager) UploadAll

func (dbm *DBManager) UploadAll(ctx context.Context) error

UploadAll uploads all db dumps in the db dir.

func (*DBManager) UploadAt

func (dbm *DBManager) UploadAt(ctx context.Context, exportPath string) error

UploadAt uploads a db dump at the given path. It returns an error if a dumps cannot be uploaded. It cleans up the db dumps and export files after uploading.

type EventInfo

type EventInfo struct {
	CID         string `json:"cid"`
	Timestamp   int64  `json:"timestamp"`
	IsArchived  bool   `json:"is_archived"`
	CacheExpiry string `json:"cache_expiry"`
}

EventInfo represents information about a deal.

type ListVaultEventsParams

type ListVaultEventsParams struct {
	Vault  Vault
	Limit  uint32
	Offset uint32
	Before Timestamp
	After  Timestamp
}

ListVaultEventsParams ...

type ListVaultsParams

type ListVaultsParams struct {
	Account *Account
}

ListVaultsParams ...

type Replicator

type Replicator interface {
	StartReplication(ctx context.Context) (chan *pgrepl.Tx, []string, error)
	Commit(ctx context.Context, lsn pglogrepl.LSN) error
	Shutdown()
}

Replicator replicates Postgres txs into a channel.

type RetrieveEventParams

type RetrieveEventParams struct {
	Timeout int64
	CID     cid.Cid
}

RetrieveEventParams ...

type Retriever

type Retriever struct {
	// contains filtered or unexported fields
}

Retriever is responsible for retrieving file from the network.

func NewRetriever

func NewRetriever(provider VaultsProvider, timeout int64) *Retriever

NewRetriever creates a new Retriever.

func (*Retriever) Retrieve

func (r *Retriever) Retrieve(ctx context.Context, c cid.Cid, output string) error

Retrieve retrieves file from the network.

type TableSchema

type TableSchema struct {
	Table   string
	Columns []Column
}

TableSchema represents a table and its schema.

type Timestamp

type Timestamp struct {
	// contains filtered or unexported fields
}

Timestamp represents a time a file was uploaded.

func NewTimestamp

func NewTimestamp(t time.Time) Timestamp

NewTimestamp creates a new Timestamp.

func ParseTimestamp

func ParseTimestamp(ts string) (Timestamp, error)

ParseTimestamp parses a string and returns a time.Time object as UTC. It accepts 3 kinds of formats: - Integers: that will be parsed as seconds - Date Only format (e.g. 2006-01-02) - RFC3339 (e.g. 2006-01-02T15:04:05Z07:00).

func (*Timestamp) Seconds

func (t *Timestamp) Seconds() int64

Seconds returns the timestamp in seconds.

type Vault

type Vault string

Vault represents a vault.

type VaultsProvider

type VaultsProvider interface {
	CreateVault(context.Context, CreateVaultParams) error
	ListVaults(context.Context, ListVaultsParams) ([]Vault, error)
	ListVaultEvents(context.Context, ListVaultEventsParams) ([]EventInfo, error)
	WriteVaultEvent(context.Context, WriteVaultEventParams) error
	RetrieveEvent(context.Context, RetrieveEventParams, io.Writer) (string, error)
}

VaultsProvider defines Vaults API.

type VaultsStreamer

type VaultsStreamer struct {
	// contains filtered or unexported fields
}

VaultsStreamer contains logic of streaming Postgres changes to Vaults Provider.

func NewVaultsStreamer

func NewVaultsStreamer(ns string, r Replicator, dbm *DBManager) *VaultsStreamer

NewVaultsStreamer creates new streamer.

func (*VaultsStreamer) Run

func (b *VaultsStreamer) Run(ctx context.Context) error

Run runs the VaultsStreamer logic.

type VaultsUploader

type VaultsUploader struct {
	// contains filtered or unexported fields
}

VaultsUploader contains logic of uploading Parquet files to Vaults Provider.

func NewVaultsUploader

func NewVaultsUploader(
	ns string, rel string, bp VaultsProvider, pk *ecdsa.PrivateKey,
) *VaultsUploader

NewVaultsUploader creates new uploader.

func (*VaultsUploader) Upload

func (bu *VaultsUploader) Upload(
	ctx context.Context, filepath string, progress io.Writer, ts Timestamp, sz int64,
) error

Upload sends file to provider for upload.

type WriteVaultEventParams

type WriteVaultEventParams struct {
	Vault       Vault
	Signature   string
	Filename    string
	Timestamp   Timestamp
	Content     io.Reader
	ProgressBar io.Writer
	Size        int64
}

WriteVaultEventParams ...