Documentation
¶
Overview ¶
Package eventlog owns the authoritative, deduplicated group-message log (ctx_group_message) and its per-group ordering registry (ctx_group_state).
Several bots may deliver the same human message; AppendGroupMessage collapses those into a single row and never silently drops one. It is the only sanctioned way to append — callers must not run raw INSERTs, or the dedup/seq invariants break.
Index ¶
- type ActorType
- type AppendOption
- type AppendResult
- type GroupMessage
- type Message
- type Store
- func (s *Store) AppendGroupMessage(ctx context.Context, msg Message, opts ...AppendOption) (AppendResult, error)
- func (s *Store) AppendToGroup(ctx context.Context, groupID string, msg GroupMessage) (AppendResult, error)
- func (s *Store) ResolveGroupID(ctx context.Context, platform, platformGroupID, platformThreadID string) (string, error)
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type ActorType ¶
type ActorType string
ActorType records who spoke. It is a hard schema fact, never guessed from content: downstream (arbiter, memory ingest) acts on human rows only.
type AppendOption ¶ added in v0.43.0
type AppendOption func(*appendConfig)
AppendOption customizes AppendGroupMessage.
func WithOnInserted ¶ added in v0.43.0
func WithOnInserted(callback func(context.Context, *sqlc.Queries, AppendResult) error) AppendOption
WithOnInserted runs callback inside AppendGroupMessage's transaction after a new row is inserted and before commit. It is not called for deduplicated redeliveries; returning an error rolls back the message insert.
type AppendResult ¶
type AppendResult struct {
GroupID string // surrogate id of the resolved registry row
Seq int64 // group-monotonic ordering token of the (existing or new) row
Inserted bool // false = an idempotent redelivery collapsed onto an existing row
Message sqlc.CtxGroupMessage
}
AppendResult reports the outcome of an append.
func AppendToGroupWithQueries ¶ added in v0.47.0
func AppendToGroupWithQueries(ctx context.Context, q *sqlc.Queries, groupID string, msg GroupMessage) (AppendResult, error)
AppendToGroupWithQueries appends to a pre-resolved group using the caller's query handle. The caller owns the surrounding transaction and MUST already hold the "gid:"+groupID advisory lock (see AppendToGroup) so this writeback stays atomic with its sibling writes — this helper lets dispatch result markers commit atomically with agent response writeback. Seq integrity itself does not depend on that lock: BumpGroupSeq's atomic UPDATE ... RETURNING guarantees a distinct seq under the row lock regardless.
type GroupMessage ¶
type GroupMessage struct {
ActorType ActorType
ActorID string
Content string
Reasoning string
AgentSessionID string
}
GroupMessage is a simplified message for direct group append (pre-resolved groupID).
type Message ¶
type Message struct {
// Physical group identity (D0). The triple resolves to one registry row.
Platform string
PlatformGroupID string
PlatformThreadID string // "" when the group has no sub-thread/topic
SourceChannelID string // observing bot; audit only, never a dedup/route key
ActorType ActorType
ActorID string // platform sender id (human) or agent id (agent)
PlatformMessageID string // "" when the adapter cannot supply one
ReplyTo string // platform message id this replies to; "" if none
// PlatformTimestamp is the platform-reported send time. Pass a non-zero,
// dedup-grade (high-precision) time to enable the fallback idempotency key
// when PlatformMessageID is absent; pass the zero value to skip it. Never
// pass a local receive time — that would misclassify back-to-back identical
// messages as redeliveries and drop data.
PlatformTimestamp time.Time
Content string
}
Message is one observed delivery to append. Content is the already-serialized payload (JSON []ai.ContentBlock by convention); eventlog treats it as opaque and stores it verbatim.
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store appends to the group event log.
func (*Store) AppendGroupMessage ¶
func (s *Store) AppendGroupMessage(ctx context.Context, msg Message, opts ...AppendOption) (AppendResult, error)
AppendGroupMessage is the single sanctioned append primitive. It runs the closed, idempotent algorithm under a per-group advisory lock (which serializes concurrent writers for the same group): resolve-or-create the registry row, check for an existing message by unique key, and only on a miss bump next_seq and insert. An idempotent redelivery neither inserts a row nor consumes a seq.
func (*Store) AppendToGroup ¶
func (s *Store) AppendToGroup(ctx context.Context, groupID string, msg GroupMessage) (AppendResult, error)
AppendToGroup appends a message directly to a pre-resolved group (bypassing triple-based group resolution). Used for agent response writeback where the groupID is already known from the dispatch flow.
It locks on "gid:"+groupID — a different namespace than AppendGroupMessage's triple key — and that split is deliberate, not a missed unification: a pre-resolved group already exists, so there is no get-or-create to serialize against the triple path, and the two paths share no mutable state except next_seq, whose allocation is already serialized by BumpGroupSeq's atomic UPDATE ... RETURNING row lock (with UNIQUE(group_id, seq) as a backstop). The advisory lock here exists to make the dispatcher's multi-statement writeback atomic, not to protect the seq bump.
func (*Store) ResolveGroupID ¶
func (s *Store) ResolveGroupID(ctx context.Context, platform, platformGroupID, platformThreadID string) (string, error)
ResolveGroupID performs a get-or-create on the group registry for the given physical (platform, group, thread) triple and returns the surrogate group_id. It runs under a per-group advisory lock so the get-or-create is atomic.
Source Files