app

Documentation

Index

• type Account
  • func NewAccount(address string) (*Account, error)
  • func (a *Account) Hex() string
• type BasinProvider
• type BasinStreamer
  • func NewBasinStreamer(ns string, r Replicator, dbm *DBManager) *BasinStreamer
  • func (b *BasinStreamer) Run(ctx context.Context) error
• type BasinUploader
  • func NewBasinUploader(ns string, rel string, bp BasinProvider, pk *ecdsa.PrivateKey) *BasinUploader
  • func (bu *BasinUploader) Upload(ctx context.Context, filepath string, progress io.Writer, ts Timestamp, ...) error
• type CacheDuration
• type Column
• type CreateVaultParams
• type DBManager
  • func NewDBManager(dbDir, table string, cols []Column, winSize time.Duration, ...) *DBManager
  • func (dbm *DBManager) Close() error
  • func (dbm *DBManager) Export(ctx context.Context, exportPath 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 Signer
  • func NewSigner(pk *ecdsa.PrivateKey) *Signer
  • func (s *Signer) Sign() ([]byte, error)
  • func (s *Signer) SignFile(filename string) ([]byte, error)
  • func (s *Signer) Sum(chunk []byte)
• type Timestamp
  • func NewTimestamp(t time.Time) Timestamp
  • func ParseTimestamp(ts string) (Timestamp, error)
  • func (t *Timestamp) Seconds() int64
• type Vault
• type WriteVaultEventParams

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 BasinProvider

type BasinProvider interface {
	CreateVault(context.Context, CreateVaultParams) error
	ListVaults(context.Context, ListVaultsParams) ([]Vault, error)
	ListVaultEvents(context.Context, ListVaultEventsParams) ([]EventInfo, error)
	WriteVaultEvent(context.Context, WriteVaultEventParams) error
}

BasinProvider ...

type BasinStreamer

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

BasinStreamer contains logic of streaming Postgres changes to Basin Provider.

func NewBasinStreamer

func NewBasinStreamer(ns string, r Replicator, dbm *DBManager) *BasinStreamer

NewBasinStreamer creates new streamer.

func (*BasinStreamer) Run

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

Run runs the BasinStreamer logic.

type BasinUploader

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

BasinUploader contains logic of uploading Parquet files to Basin Provider.

func NewBasinUploader

func NewBasinUploader(
	ns string, rel string, bp BasinProvider, pk *ecdsa.PrivateKey,
) *BasinUploader

NewBasinUploader creates new uploader.

func (*BasinUploader) Upload

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

Upload sends file to provider for upload.

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, table string, cols []Column, winSize time.Duration, uploader *BasinUploader) *DBManager

NewDBManager creates a new DBManager.

func (*DBManager) Close

func (dbm *DBManager) Close() error

Close closes the current db.

func (*DBManager) Export

func (dbm *DBManager) Export(ctx context.Context, exportPath 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"`
	Size        uint32 `json:"size"`
	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 Signer

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

Signer allows you to sign a big stream of bytes by calling Sum multiple times, then Sign.

func NewSigner

func NewSigner(pk *ecdsa.PrivateKey) *Signer

NewSigner creates a new signer.

func (*Signer) Sign

func (s *Signer) Sign() ([]byte, error)

Sign signs the internal state.

func (*Signer) SignFile

func (s *Signer) SignFile(filename string) ([]byte, error)

SignFile signs an entire file.

func (*Signer) Sum

func (s *Signer) Sum(chunk []byte)

Sum updates the hash state with a new chunk.

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 WriteVaultEventParams

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

WriteVaultEventParams ...