spool

type Applier interface {
	// EnsureSchema creates whatever bookkeeping the applier needs, once, before any
	// segment is written.
	EnsureSchema(ctx context.Context) error

	// AlreadyApplied reports a segment recovery found on disk that the database already
	// holds, so it can be dropped rather than replayed.
	//
	// A driver with transactions answers this from its own record of applied segments. One
	// without answers it from how far the stored cursor got: replaying a segment it had in
	// fact applied duplicates exactly the rows re-streaming it would have duplicated, which
	// is the guarantee such a driver already ships with.
	AlreadyApplied(ctx context.Context, manifest *Manifest) (bool, error)

	// Apply loads one segment and advances the cursor with it.
	Apply(ctx context.Context, dir string, manifest *Manifest) error
}

type Codec interface {
	// Format is what gets recorded in the manifest.
	Format() Format

	// OpenSegment prepares the writers for a new segment in dir, which already exists.
	OpenSegment(dir string) (SegmentWriter, error)

	// Verify checks a sealed segment's files against its manifest. Only the manifest is
	// fsynced on seal, so a machine crash can leave a data file short and this is what
	// catches it before anything reaches the database.
	Verify(dir string, manifest *Manifest) error
}

type Format string

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

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

type Manifest struct {
	FirstBlock uint64        `json:"first_block"`
	LastBlock  uint64        `json:"last_block"`
	Cursor     string        `json:"cursor"`
	Tables     []TableRecord `json:"tables"`
	Sealed     bool          `json:"sealed"`

	// Format says how the rows are laid out. Absent means FormatPGCopy, which is what
	// every segment written before formats existed carries.
	Format Format `json:"format,omitempty"`
	// LogFile and LogBytes describe the single interleaved file of FormatRowLog. The
	// table records still carry the column layout, since replaying a row needs it.
	LogFile  string `json:"log_file,omitempty"`
	LogBytes int64  `json:"log_bytes,omitempty"`
}

type Options struct {
	// Dir is where segments live. Required.
	Dir string
	// MaxBytes is the disk quota, and the only bound on how far ahead of the database the
	// stream may run. Writes are held once the spool holds this much waiting to be
	// applied, which is what turns a slow database into backpressure rather than a full
	// disk. Zero picks 8GiB.
	MaxBytes int64
	// WriteTargetDuration is how long one commit to the database should take. The sizer
	// measures each commit and steers the next segment toward it. Zero picks 3s.
	WriteTargetDuration time.Duration
	// SegmentMaxBytes is the ceiling the sizer may choose, whatever the target duration
	// would allow. Zero picks 512MiB. The floor is segmentFloorBytes and not
	// configurable, see sizer.go.
	SegmentMaxBytes int64
	// MaxIdle commits the open segment once no new row has reached it for this long, short
	// of its size target. Without it a stalled stream sits on those rows indefinitely,
	// leaving the cursor where it was. Zero picks 10s; negative disables idle sealing.
	MaxIdle time.Duration
}

type SealReason int

type SegmentWriter interface {
	// WriteRow appends one row of the named table.
	WriteRow(table string, values []any) error

	// PendingBytes is how much has been written so far, from the writers' own counters
	// rather than a stat() per call. It is what the sizer compares against.
	PendingBytes() int64

	// Seal closes every stream and fills in the manifest's file records.
	Seal(manifest *Manifest) error

	// Discard closes and abandons a segment that will never be applied.
	Discard()
}

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

type Stats struct {
	// Applied, cumulative: each counts only what a segment that reached the database
	// carried, never what is still queued. Bytes is that segment's size on disk, in
	// whatever format the codec writes — what the sizer steers and the disk quota bounds,
	// not what crossed the network.
	Segments      int64
	Blocks        int64
	Rows          int64
	Bytes         int64
	ApplyDuration time.Duration

	// ApplierBusy and ApplierIdle split the applier goroutine's wall clock. Their ratio
	// is the answer to "is the database or the stream the limit": an applier that is busy
	// essentially all the time is the ceiling, one that mostly waits is not.
	ApplierBusy time.Duration
	ApplierIdle time.Duration

	// QuotaWait is how long the stream was held because the spool was full. Any non-zero
	// value means the database is gating download.
	QuotaWait time.Duration

	// Sealed counts segments by what closed them, indexed by SealReason.
	Sealed [sealReasonCount]int64

	// In flight right now.
	QueueDepth    int
	SegmentTarget int64
	OpenBytes     int64
	OpenBlocks    int64
	OpenAge       time.Duration

	// Queued totals, sealed segments plus the one being written.
	BlocksBuffered int64
	BytesOnDisk    int64
	AppliedBlock   uint64
	Quota          int64
}

type TableRecord struct {
	Name     string   `json:"name"`
	Schema   string   `json:"schema"`
	Relation string   `json:"relation"`
	File     string   `json:"file"`
	Columns  []string `json:"columns"`
	Rows     int64    `json:"rows"`
	Bytes    int64    `json:"bytes"`
}