53 lines
1.5 KiB
Go
53 lines
1.5 KiB
Go
package ingest
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"io"
|
|
)
|
|
|
|
// Sink materializes a stream into the reporting database.
|
|
//
|
|
// Idempotency lives HERE and not in the reader, because the effect and the high-water mark
|
|
// (run_attempts.last_seq) have to move in ONE transaction: an implementation that applies an event
|
|
// and then records it has a crash window that duplicates work.
|
|
//
|
|
// This is a reporting database, not an event store (research/23 §3): the events are not kept.
|
|
type Sink interface {
|
|
// Begin binds the stream to an attempt — the engine's run id is the idempotency namespace.
|
|
Begin(ctx context.Context, h Hello) error
|
|
// Apply materializes one event. It MUST ignore an event whose Seq is not greater than the
|
|
// stored high-water mark, and it MUST ignore an unknown Type.
|
|
Apply(ctx context.Context, ev Envelope) error
|
|
}
|
|
|
|
// Ingest reads a stream to its end, feeding a sink.
|
|
//
|
|
// It stops at the first error and returns it: a gap or a malformed line is not recoverable by
|
|
// reading further — the caller reconciles from `tmctl status --json`, which is the whole reason
|
|
// that channel is part of the ratified seam.
|
|
func Ingest(ctx context.Context, r io.Reader, sink Sink) error {
|
|
d := NewDecoder(r)
|
|
h, err := d.Hello()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := sink.Begin(ctx, h); err != nil {
|
|
return err
|
|
}
|
|
for {
|
|
if err := ctx.Err(); err != nil {
|
|
return err
|
|
}
|
|
ev, err := d.Next()
|
|
if errors.Is(err, io.EOF) {
|
|
return nil
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := sink.Apply(ctx, ev); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|