auditforward

type Chain struct {
	// Source is SourceControlPlane or SourceProject.
	Source string
	// Dir is the directory holding .cloop/state.db.
	Dir string
	// Path is the database's absolute path: what a cursor is keyed on.
	Path string
}

type ChainReport struct {
	Cursor statedb.AuditForwardCursor
	// HeadID is the chain's newest row; Backlog how many rows after the
	// cursor the sink has not acknowledged; PendingSince the oldest of
	// those's timestamp, zero when there are none.
	HeadID       int64
	Backlog      int64
	PendingSince time.Time
	// Err is set when the chain could not be read.
	Err error
}

type ChainStatus struct {
	Chain Chain
	// AckedID is the newest row the sink acknowledged; HeadID the newest
	// row the chain held when last read.
	AckedID int64
	HeadID  int64
	// PendingSince is the timestamp of the oldest row the sink has not
	// acknowledged; zero when it is caught up.
	PendingSince time.Time
}

type ChainVerdict struct {
	Source string   `json:"source"`
	Dir    string   `json:"dir"`
	Hubs   []string `json:"hubs,omitempty"`

	FirstID int64 `json:"first_id"`
	LastID  int64 `json:"last_id"`
	// Rows counts distinct rows; Duplicates the records that repeated one.
	Rows       int `json:"rows"`
	Duplicates int `json:"duplicates"`

	// Gaps are ids between FirstID and LastID the stream does not hold.
	Gaps []IDRange `json:"gaps,omitempty"`
	// Forks are ids held with more than one row_hash.
	Forks []int64 `json:"forks,omitempty"`
	// BadHashes are records whose content does not hash to their row_hash.
	BadHashes []int64 `json:"bad_hashes,omitempty"`
	// BrokenLinks are records whose prev_hash is not the row_hash of the
	// record before them — or, for id 1, not genesis.
	BrokenLinks []int64 `json:"broken_links,omitempty"`
	// Unverified are records sent with their payload left out, whose hash
	// cannot be recomputed; their links are still checked.
	Unverified []int64 `json:"unverified,omitempty"`
}

type Credentials func(ref string) ([]byte, error)

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

type Hooks struct {
	// AfterAck runs after a sink acknowledged a batch and before its cursor
	// moves. A test that never returns from it, or calls runtime.Goexit, is a
	// leader killed between the two.
	AfterAck func(sink string, chain Chain, firstID, lastID int64)
}

type IDRange struct {
	From int64 `json:"from"`
	To   int64 `json:"to"`
}

type Options struct {
	// ControlPlane is the control plane's database path: where the cursors
	// live and where the forwarder's own audit rows go.
	ControlPlane string

	// Config returns the audit.forward section in force. Re-read every
	// ConfigPoll, so an edit takes effect without a restart.
	Config func() (config.AuditForwardConfig, error)

	// ConfigSource names where Config read from, for the audit row recording
	// a change ("config.yaml", or the overlay merged over it).
	ConfigSource string

	// Refused lists the sinks the loader switched off, by repair key, for the
	// same audit row. May be nil.
	Refused func() []string

	// Chains returns the chains to forward, each database once, the control
	// plane first. Re-read every ConfigPoll.
	Chains func() []Chain

	// Credentials opens the secrets sinks name. Nil makes every sink that
	// names one fail to start, with that said.
	Credentials Credentials

	// Version is stamped into CEF records and the HTTP User-Agent.
	Version string

	// Poll is how often a caught-up sink looks for new rows. Zero means two
	// seconds.
	Poll time.Duration

	// ConfigPoll is how often the configuration and the chain list are
	// re-read. Zero means ten seconds.
	ConfigPoll time.Duration

	// Logf receives one line per noteworthy event: a sink failing, a
	// discontinuity. Nil writes to stderr.
	Logf func(format string, args ...any)

	// Now overrides the clock. Nil means time.Now.
	Now func() time.Time

	// Hooks are for tests.
	Hooks Hooks
}

type Report struct {
	Sink   config.AuditForwardSink
	Chains []ChainReport
}

type SinkDescription struct {
	Name       string   `json:"name"`
	Type       string   `json:"type"`
	Format     string   `json:"format"`
	Endpoint   string   `json:"endpoint"`
	Variant    string   `json:"variant,omitempty"`
	Refs       []string `json:"refs,omitempty"`
	CAFile     string   `json:"ca_file,omitempty"`
	ClientCert string   `json:"client_cert,omitempty"`
	ServerName string   `json:"server_name,omitempty"`
	// PlaintextLoopback is true for a sink acknowledged to send plaintext to
	// a loopback collector.
	PlaintextLoopback bool   `json:"plaintext_loopback,omitempty"`
	Start             string `json:"start"`
	Fingerprint       string `json:"fingerprint"`
}

type SinkStatus struct {
	Name     string
	Type     string
	Endpoint string

	// Chains is per chain, keyed by database path.
	Chains map[string]ChainStatus

	// LastError is the newest send failure, cleared by a delivery.
	LastError   string
	LastErrorAt time.Time
	// LastDeliveredAt is when the sink last acknowledged a batch.
	LastDeliveredAt time.Time
}

type StreamReport struct {
	Chains []ChainVerdict `json:"chains"`
	// Lines counts the lines read; Skipped those that held no record, and
	// Malformed those that started one and did not decode.
	Lines     int   `json:"lines"`
	Skipped   int   `json:"skipped"`
	Malformed []int `json:"malformed,omitempty"`
}