weichenandCursor
d5968ad4ba
Workers claim with SKIP LOCKED, heartbeat by token, and ack/fail only when the fencing token still matches. Shadow scans without acking live deliveries. Co-authored-by: Cursor <cursoragent@cursor.com>
71 lines
1.9 KiB
Go
71 lines
1.9 KiB
Go
package events
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"time"
|
|
|
|
"chat-agent-system/internal/app/uow"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
// PGSink appends envelopes inside the current UnitOfWork.
|
|
type PGSink struct{}
|
|
|
|
func (PGSink) Append(ctx context.Context, env Envelope, consumers []string) error {
|
|
if err := AssertLiveConsumers(env.EventName, consumers); err != nil {
|
|
return err
|
|
}
|
|
scope, ok := uow.ScopeFrom(ctx)
|
|
if !ok {
|
|
return fmt.Errorf("events: postgres sink requires a unit of work")
|
|
}
|
|
tx, err := uow.AsPG(scope)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return InsertEnvelope(ctx, tx, env, consumers)
|
|
}
|
|
|
|
func InsertEnvelope(ctx context.Context, tx pgx.Tx, env Envelope, consumers []string) error {
|
|
if err := ValidateEnvelope(env); err != nil {
|
|
return err
|
|
}
|
|
_, err := tx.Exec(ctx, `
|
|
INSERT INTO event_envelopes (
|
|
event_id, event_name, schema_version, environment, project_id,
|
|
aggregate_type, aggregate_id, aggregate_version, actor_ref,
|
|
correlation_id, causation_id, idempotency_key, payload_digest,
|
|
producer, occurred_at, payload
|
|
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16)
|
|
ON CONFLICT (event_id) DO NOTHING
|
|
`, env.EventID, env.EventName, env.SchemaVersion, env.Environment, env.ProjectID,
|
|
env.AggregateType, env.AggregateID, env.AggregateVersion, env.ActorRef,
|
|
env.CorrelationID, env.CausationID, env.IdempotencyKey, env.PayloadDigest,
|
|
env.Producer, env.OccurredAt, env.Payload)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for _, c := range consumers {
|
|
if _, err := tx.Exec(ctx, `
|
|
INSERT INTO event_deliveries (event_id, consumer, status, next_attempt_at)
|
|
VALUES ($1,$2,'pending', now())
|
|
ON CONFLICT DO NOTHING
|
|
`, env.EventID, c); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func PingPool(ctx context.Context, pool *pgxpool.Pool) error {
|
|
if pool == nil {
|
|
return ErrUnknownConsumer
|
|
}
|
|
c, cancel := context.WithTimeout(ctx, 2*time.Second)
|
|
defer cancel()
|
|
return pool.Ping(c)
|
|
}
|