452 lines
14 KiB
Go
452 lines
14 KiB
Go
package events
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"chat-agent-system/internal/errcode"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
const (
|
|
DefaultBatch = 50
|
|
DefaultLease = 30 * time.Second
|
|
DefaultHeartbeat = 10 * time.Second
|
|
DefaultBackoffCeiling = 15 * time.Minute
|
|
ModeOff = "off"
|
|
ModeShadow = "shadow"
|
|
ModeLive = "live"
|
|
)
|
|
|
|
// Lease is the fenced claim a worker holds while the handler runs outside the claim transaction.
|
|
type Lease struct {
|
|
EventID string
|
|
Consumer string
|
|
Attempt int
|
|
Owner string
|
|
Token string
|
|
LeaseUntil time.Time
|
|
AggregateType string
|
|
AggregateID string
|
|
ProjectID int64
|
|
Environment string
|
|
EventName string
|
|
Envelope Envelope
|
|
}
|
|
|
|
// VersionClaimable is the gap rule shared by MemoryBus and claimSQL.
|
|
// Version 0 is unversioned and always claimable. A versioned event is claimable
|
|
// only when it does not skip past the consumer checkpoint (head+1).
|
|
func VersionClaimable(head, version int64) bool {
|
|
if version == 0 {
|
|
return true
|
|
}
|
|
return version <= head+1
|
|
}
|
|
|
|
const claimSQL = `
|
|
WITH candidates AS (
|
|
SELECT d.event_id, d.consumer
|
|
FROM event_deliveries d
|
|
INNER JOIN event_envelopes e ON e.event_id = d.event_id
|
|
WHERE d.consumer = $1
|
|
AND d.consumer <> 'fav'
|
|
AND d.status IN ('pending','leased')
|
|
AND (d.lease_until IS NULL OR d.lease_until < now())
|
|
AND d.next_attempt_at <= now()
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM event_inventory_decisions inv
|
|
WHERE inv.event_id = d.event_id
|
|
AND (inv.consumer = d.consumer OR inv.consumer = '')
|
|
AND inv.classification IN ('quarantine','expire')
|
|
)
|
|
AND NOT EXISTS (
|
|
SELECT 1
|
|
FROM event_deliveries x
|
|
JOIN event_envelopes ex ON ex.event_id = x.event_id
|
|
WHERE x.consumer = d.consumer
|
|
AND x.status = 'leased'
|
|
AND x.lease_until > now()
|
|
AND ex.aggregate_type = e.aggregate_type
|
|
AND ex.aggregate_id = e.aggregate_id
|
|
)
|
|
AND (
|
|
e.aggregate_version = 0
|
|
OR e.aggregate_version <= 1 + COALESCE((
|
|
SELECT c.last_aggregate_version
|
|
FROM event_consumer_checkpoints c
|
|
WHERE c.consumer = d.consumer
|
|
AND c.aggregate_type = e.aggregate_type
|
|
AND c.aggregate_id = e.aggregate_id
|
|
), 0)
|
|
)
|
|
ORDER BY d.next_attempt_at, d.event_id
|
|
FOR UPDATE OF d SKIP LOCKED
|
|
LIMIT $2
|
|
)
|
|
UPDATE event_deliveries d
|
|
SET status = 'leased',
|
|
lease_owner = $3,
|
|
fencing_token = md5(random()::text || clock_timestamp()::text || d.event_id),
|
|
lease_until = now() + $4::interval,
|
|
attempt = d.attempt + 1
|
|
FROM candidates c
|
|
WHERE d.event_id = c.event_id AND d.consumer = c.consumer
|
|
RETURNING d.event_id, d.consumer, d.attempt, d.lease_owner, d.fencing_token, d.lease_until
|
|
`
|
|
|
|
// ClaimIDs is the retired SELECT-only path. Production workers must use ClaimLeases.
|
|
func ClaimIDs(ctx context.Context, tx pgx.Tx, consumer string, limit int) ([]string, error) {
|
|
return nil, fmt.Errorf("events: SELECT-only claim is disabled; use ClaimLeases")
|
|
}
|
|
|
|
const peekSQL = `
|
|
SELECT d.event_id, d.consumer, d.attempt
|
|
FROM event_deliveries d
|
|
INNER JOIN event_envelopes e ON e.event_id = d.event_id
|
|
WHERE d.consumer = $1
|
|
AND d.consumer <> 'fav'
|
|
AND d.status IN ('pending','leased')
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM event_inventory_decisions inv
|
|
WHERE inv.event_id = d.event_id
|
|
AND (inv.consumer = d.consumer OR inv.consumer = '')
|
|
AND inv.classification IN ('quarantine','expire')
|
|
)
|
|
ORDER BY d.next_attempt_at, d.event_id
|
|
LIMIT $2
|
|
`
|
|
|
|
// PeekPendingLeases hydrates deliveries without claiming. Shadow uses this so
|
|
// live fencing tokens and attempt rows stay untouched.
|
|
func PeekPendingLeases(ctx context.Context, pool *pgxpool.Pool, consumer string, limit int) ([]Lease, error) {
|
|
if consumer == ConsumerFAV || strings.EqualFold(consumer, "fav") {
|
|
return nil, fmt.Errorf("%w: fav", ErrDisabledConsumer)
|
|
}
|
|
if pool == nil {
|
|
return nil, fmt.Errorf("events: peek requires pool")
|
|
}
|
|
if limit <= 0 {
|
|
limit = DefaultBatch
|
|
}
|
|
rows, err := pool.Query(ctx, peekSQL, consumer, limit)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var out []Lease
|
|
for rows.Next() {
|
|
var l Lease
|
|
if err := rows.Scan(&l.EventID, &l.Consumer, &l.Attempt); err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, l)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
tx, err := pool.Begin(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer func() { _ = tx.Rollback(ctx) }()
|
|
for i := range out {
|
|
if err := hydrateLease(ctx, tx, &out[i]); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func ClaimLeases(ctx context.Context, tx pgx.Tx, consumer, owner string, limit int, lease time.Duration) ([]Lease, error) {
|
|
if consumer == ConsumerFAV || strings.EqualFold(consumer, "fav") {
|
|
return nil, fmt.Errorf("%w: fav", ErrDisabledConsumer)
|
|
}
|
|
if limit <= 0 {
|
|
limit = DefaultBatch
|
|
}
|
|
if lease <= 0 {
|
|
lease = DefaultLease
|
|
}
|
|
rows, err := tx.Query(ctx, claimSQL, consumer, limit, owner, fmt.Sprintf("%d seconds", int(lease.Seconds())))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var out []Lease
|
|
for rows.Next() {
|
|
var l Lease
|
|
if err := rows.Scan(&l.EventID, &l.Consumer, &l.Attempt, &l.Owner, &l.Token, &l.LeaseUntil); err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, l)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
for i := range out {
|
|
if _, err := tx.Exec(ctx, `
|
|
INSERT INTO event_delivery_attempts (
|
|
event_id, consumer, attempt, lease_owner, fencing_token, outcome, correlation_id
|
|
) VALUES ($1,$2,$3,$4,$5,'leased',$1)
|
|
`, out[i].EventID, out[i].Consumer, out[i].Attempt, out[i].Owner, out[i].Token); err != nil {
|
|
return nil, err
|
|
}
|
|
if err := hydrateLease(ctx, tx, &out[i]); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func hydrateLease(ctx context.Context, tx pgx.Tx, l *Lease) error {
|
|
var env Envelope
|
|
if err := tx.QueryRow(ctx, `
|
|
SELECT event_id, event_name, schema_version, environment, COALESCE(project_id,0),
|
|
aggregate_type, aggregate_id, aggregate_version, actor_ref,
|
|
correlation_id, causation_id, idempotency_key, payload_digest,
|
|
producer, occurred_at, payload
|
|
FROM event_envelopes WHERE event_id=$1
|
|
`, l.EventID).Scan(&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); err != nil {
|
|
return err
|
|
}
|
|
l.Envelope = env
|
|
l.AggregateType = env.AggregateType
|
|
l.AggregateID = env.AggregateID
|
|
l.ProjectID = env.ProjectID
|
|
l.Environment = env.Environment
|
|
l.EventName = env.EventName
|
|
return nil
|
|
}
|
|
|
|
func AckLease(ctx context.Context, tx pgx.Tx, l Lease) error {
|
|
tag, err := tx.Exec(ctx, `
|
|
UPDATE event_deliveries
|
|
SET status='acked', lease_owner='', fencing_token='', inbox=TRUE
|
|
WHERE event_id=$1 AND consumer=$2 AND lease_owner=$3 AND fencing_token=$4 AND status='leased'
|
|
`, l.EventID, l.Consumer, l.Owner, l.Token)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if tag.RowsAffected() != 1 {
|
|
return ErrStaleFencing
|
|
}
|
|
if _, err := tx.Exec(ctx, `
|
|
UPDATE event_delivery_attempts
|
|
SET finished_at=now(), outcome='acked'
|
|
WHERE event_id=$1 AND consumer=$2 AND fencing_token=$3
|
|
`, l.EventID, l.Consumer, l.Token); err != nil {
|
|
return err
|
|
}
|
|
if l.AggregateType != "" && l.AggregateID != "" {
|
|
if _, err := tx.Exec(ctx, `
|
|
INSERT INTO event_consumer_checkpoints (
|
|
consumer, aggregate_type, aggregate_id, last_event_id, last_aggregate_version, fencing_token
|
|
) VALUES ($1,$2,$3,$4,$5,$6)
|
|
ON CONFLICT (consumer, aggregate_type, aggregate_id) DO UPDATE SET
|
|
last_event_id = EXCLUDED.last_event_id,
|
|
last_aggregate_version = GREATEST(COALESCE(event_consumer_checkpoints.last_aggregate_version, 0), EXCLUDED.last_aggregate_version),
|
|
fencing_token = EXCLUDED.fencing_token,
|
|
updated_at = now()
|
|
`, l.Consumer, l.AggregateType, l.AggregateID, l.EventID, l.Envelope.AggregateVersion, l.Token); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func FailLease(ctx context.Context, tx pgx.Tx, l Lease, code string, retryable bool) error {
|
|
nextStatus := "pending"
|
|
outcome := "pending"
|
|
if !retryable || l.Attempt >= MaxAttempts {
|
|
nextStatus = "dead"
|
|
outcome = "dead"
|
|
if code == "QUARANTINE" || code == "AUTHZ" || code == "SCOPE" {
|
|
nextStatus = "quarantine"
|
|
outcome = "quarantine"
|
|
}
|
|
}
|
|
backoff := backoffFor(l.Attempt)
|
|
tag, err := tx.Exec(ctx, `
|
|
UPDATE event_deliveries
|
|
SET status=$5, lease_owner='', fencing_token='',
|
|
next_attempt_at = now() + $6::interval
|
|
WHERE event_id=$1 AND consumer=$2 AND lease_owner=$3 AND fencing_token=$4 AND status='leased'
|
|
`, l.EventID, l.Consumer, l.Owner, l.Token, nextStatus, fmt.Sprintf("%d seconds", int(backoff.Seconds())))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if tag.RowsAffected() != 1 {
|
|
return ErrStaleFencing
|
|
}
|
|
_, err = tx.Exec(ctx, `
|
|
UPDATE event_delivery_attempts
|
|
SET finished_at=now(), outcome=$4, error_code=$5
|
|
WHERE event_id=$1 AND consumer=$2 AND fencing_token=$3
|
|
`, l.EventID, l.Consumer, l.Token, outcome, code)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if outcome == "dead" || outcome == "quarantine" {
|
|
id := "EVENTS_DEAD"
|
|
if outcome == "quarantine" {
|
|
id = "EVENTS_QUARANTINE"
|
|
}
|
|
errcode.Report(ctx, errcode.Event{
|
|
Code: id,
|
|
Module: "events",
|
|
Source: "async",
|
|
RequestID: l.EventID,
|
|
PhaseID: l.Consumer,
|
|
Upstream: code,
|
|
HTTPStatus: 500,
|
|
})
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func HeartbeatLease(ctx context.Context, tx pgx.Tx, l Lease, lease time.Duration) error {
|
|
if lease <= 0 {
|
|
lease = DefaultLease
|
|
}
|
|
tag, err := tx.Exec(ctx, `
|
|
UPDATE event_deliveries
|
|
SET lease_until = now() + $5::interval
|
|
WHERE event_id=$1 AND consumer=$2 AND lease_owner=$3 AND fencing_token=$4 AND status='leased'
|
|
`, l.EventID, l.Consumer, l.Owner, l.Token, fmt.Sprintf("%d seconds", int(lease.Seconds())))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if tag.RowsAffected() != 1 {
|
|
return ErrStaleFencing
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func LoadMode(ctx context.Context, pool *pgxpool.Pool, environment, consumer string) (string, int64, error) {
|
|
var mode string
|
|
var rev int64
|
|
err := pool.QueryRow(ctx, `
|
|
SELECT mode, revision FROM event_consumer_modes WHERE environment=$1 AND consumer=$2
|
|
`, environment, consumer).Scan(&mode, &rev)
|
|
if err == pgx.ErrNoRows {
|
|
return ModeOff, 0, nil
|
|
}
|
|
return mode, rev, err
|
|
}
|
|
|
|
// WriterLive is true only when the durable consumer already owns the visible write.
|
|
// Closures must no-op in that window; off/shadow keep the in-process hook as the unique writer.
|
|
// A missing pool or mode row is treated as not-live so HTTP notify/advance keep working.
|
|
func WriterLive(ctx context.Context, pool *pgxpool.Pool, environment, consumer string) bool {
|
|
if pool == nil {
|
|
return false
|
|
}
|
|
mode, _, err := LoadMode(ctx, pool, environment, consumer)
|
|
return err == nil && mode == ModeLive
|
|
}
|
|
|
|
func SetMode(ctx context.Context, pool *pgxpool.Pool, environment, consumer, mode string) error {
|
|
if consumer == ConsumerFAV && mode != ModeOff {
|
|
return fmt.Errorf("%w: fav", ErrDisabledConsumer)
|
|
}
|
|
_, err := pool.Exec(ctx, `
|
|
INSERT INTO event_consumer_modes (environment, consumer, mode, revision, updated_at)
|
|
VALUES ($1,$2,$3,1,now())
|
|
ON CONFLICT (environment, consumer) DO UPDATE SET
|
|
mode = EXCLUDED.mode,
|
|
revision = event_consumer_modes.revision + 1,
|
|
updated_at = now()
|
|
`, environment, consumer, mode)
|
|
return err
|
|
}
|
|
|
|
func backoffFor(attempt int) time.Duration {
|
|
shift := attempt
|
|
if shift < 0 {
|
|
shift = 0
|
|
}
|
|
if shift > 10 {
|
|
shift = 10
|
|
}
|
|
d := time.Second * time.Duration(1<<shift)
|
|
if d > DefaultBackoffCeiling {
|
|
return DefaultBackoffCeiling
|
|
}
|
|
return d
|
|
}
|
|
|
|
func SchemaReady(ctx context.Context, pool *pgxpool.Pool) error {
|
|
if pool == nil {
|
|
return ErrUnknownConsumer
|
|
}
|
|
var n int
|
|
if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM event_consumer_modes`).Scan(&n); err != nil {
|
|
return err
|
|
}
|
|
if n == 0 {
|
|
return fmt.Errorf("events: consumer mode table empty")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func WriteReconciliation(ctx context.Context, tx pgx.Tx, eventID, consumer, mode, digest string) error {
|
|
_, err := tx.Exec(ctx, `
|
|
INSERT INTO event_reconciliations (event_id, consumer, mode, intended_digest)
|
|
VALUES ($1,$2,$3,$4)
|
|
ON CONFLICT (event_id, consumer, mode) DO UPDATE SET intended_digest = EXCLUDED.intended_digest
|
|
`, eventID, consumer, mode, digest)
|
|
return err
|
|
}
|
|
|
|
// VisibleOutcomeDigest is the canonical notify comparison key: recipient, type, dedupe.
|
|
func VisibleOutcomeDigest(eventID, notifType string, userID int64) string {
|
|
return Digest([]byte(visibleOutcomeLine(eventID, notifType, userID)))
|
|
}
|
|
|
|
func visibleOutcomeLine(eventID, notifType string, userID int64) string {
|
|
key := fmt.Sprintf("evt:%s:%s:%d", eventID, notifType, userID)
|
|
return fmt.Sprintf("%d|%s|%s", userID, notifType, key)
|
|
}
|
|
|
|
func EmptyVisibleDigest() string {
|
|
return Digest([]byte("empty"))
|
|
}
|
|
|
|
// HashVisibleNotifications fingerprints inbox rows on the current (possibly uncommitted) tx.
|
|
func HashVisibleNotifications(ctx context.Context, tx pgx.Tx, eventID string) (string, error) {
|
|
rows, err := tx.Query(ctx, `
|
|
SELECT user_id, type, COALESCE(dedupe_key, '')
|
|
FROM notifications
|
|
WHERE source_event_id = $1
|
|
ORDER BY user_id, type, dedupe_key
|
|
`, eventID)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
defer rows.Close()
|
|
var parts []string
|
|
for rows.Next() {
|
|
var userID int64
|
|
var typ, key string
|
|
if err := rows.Scan(&userID, &typ, &key); err != nil {
|
|
return "", err
|
|
}
|
|
parts = append(parts, fmt.Sprintf("%d|%s|%s", userID, typ, key))
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return "", err
|
|
}
|
|
if len(parts) == 0 {
|
|
return EmptyVisibleDigest(), nil
|
|
}
|
|
return Digest([]byte(strings.Join(parts, "\n"))), nil
|
|
}
|