1package daemon23import (4 "context"5 "database/sql"6 "encoding/json"7 "errors"89 "github.com/FelineStateMachine/tinyrelay/internal/event"10 "github.com/FelineStateMachine/tinyrelay/internal/replication"11 "github.com/FelineStateMachine/tinyrelay/internal/storage"12)1314type replicationPushFollowupPayload struct {15 Event event.Event `json:"event"`16 AcceptedAt int64 `json:"accepted_at"`17 RegistrationIDs []string `json:"registration_ids,omitempty"`18 RegistrationCutoff int64 `json:"registration_cutoff,omitempty"`19 Discover bool `json:"discover,omitempty"`20}2122// prepareReplicationPushFollowup captures registrations that existed when the23// source event was accepted. RegistrationCutoff is the SQLite insertion24// sequence, rather than the registration event's authored timestamp; clients25// can author addressable registrations with arbitrary timestamps. A discovery26// error is represented in the payload so the shared event-followup worker can27// retry it after the source event commits.28func (t *Tenant) prepareReplicationPushFollowup(ctx context.Context, e event.Event, now int64) (replicationPushFollowupPayload, error) {29 payload := replicationPushFollowupPayload{Event: e, AcceptedAt: now}30 ids, cutoff, err := t.captureReplicationPushRegistrations(ctx, e)31 payload.RegistrationIDs, payload.RegistrationCutoff, payload.Discover = ids, cutoff, err != nil32 return payload, nil33}3435// recoverReplicationPushFollowup discovers only registrations inserted no36// later than the acceptance cutoff. This prevents a registration created while37// the worker was retrying from receiving an older event.38func (t *Tenant) recoverReplicationPushFollowup(ctx context.Context, payload *replicationPushFollowupPayload) error {39 if !payload.Discover {40 return nil41 }42 ids, err := t.discoverReplicationPushRegistrations(ctx, payload.Event, payload.RegistrationCutoff)43 if err != nil {44 return err45 }46 payload.RegistrationIDs = appendUniqueStrings(payload.RegistrationIDs, ids...)47 payload.Discover = false48 return nil49}5051func (t *Tenant) prepareReplicationPushIntents(ctx context.Context, payload replicationPushFollowupPayload) ([]storage.Intent, error) {52 registrations, err := t.loadReplicationPushRegistrations(ctx, payload.RegistrationIDs, payload.RegistrationCutoff)53 if err != nil {54 return nil, err55 }56 return replication.PrepareCallbackIntents(ctx, payload.Event, registrations), nil57}5859func (t *Tenant) planReplicationPushFollowup(ctx context.Context, payload replicationPushFollowupPayload) ([]storage.Intent, error) {60 if !t.Policy().Features.Push {61 return nil, nil62 }63 if err := t.recoverReplicationPushFollowup(ctx, &payload); err != nil {64 return nil, err65 }66 return t.prepareReplicationPushIntents(ctx, payload)67}6869func (t *Tenant) captureReplicationPushRegistrations(ctx context.Context, e event.Event) ([]string, int64, error) {70 var cutoff int6471 if err := t.store.DB().QueryRowContext(ctx, "SELECT COALESCE(MAX(seq),0) FROM events").Scan(&cutoff); err != nil {72 return nil, 0, err73 }74 return t.discoverReplicationPushRegistrationsAt(ctx, e, cutoff)75}7677func (t *Tenant) discoverReplicationPushRegistrations(ctx context.Context, e event.Event, cutoff int64) ([]string, error) {78 if cutoff <= 0 {79 return nil, nil80 }81 ids, _, err := t.discoverReplicationPushRegistrationsAt(ctx, e, cutoff)82 return ids, err83}8485func (t *Tenant) discoverReplicationPushRegistrationsAt(ctx context.Context, e event.Event, cutoff int64) ([]string, int64, error) {86 rows, err := t.store.DB().QueryContext(ctx, `SELECT seq,id,raw FROM events WHERE kind=? AND seq<=? ORDER BY seq`, event.KIND_PUSH_REGISTRATION, cutoff)87 if err != nil {88 return nil, cutoff, err89 }90 defer rows.Close()91 approved := t.callbackPolicy()92 ids := []string{}93 for rows.Next() {94 var seq int6495 var id, raw string96 if err := rows.Scan(&seq, &id, &raw); err != nil {97 return ids, cutoff, err98 }99 var registration event.Event100 if json.Unmarshal([]byte(raw), ®istration) != nil {101 continue102 }103 parsed, parseErr := replication.ParsePushRegistration(registration, approved, t.RelayURL())104 if parseErr == nil && replication.PushMatches(parsed, e) {105 ids = append(ids, id)106 }107 }108 return ids, cutoff, rows.Err()109}110111func (t *Tenant) loadReplicationPushRegistrations(ctx context.Context, ids []string, cutoff int64) ([]replication.PushRegistration, error) {112 approved := t.callbackPolicy()113 registrations := make([]replication.PushRegistration, 0, len(ids))114 for _, id := range ids {115 var seq int64116 var raw string117 err := t.store.DB().QueryRowContext(ctx, `SELECT seq,raw FROM events WHERE id=? AND kind=?`, id, event.KIND_PUSH_REGISTRATION).Scan(&seq, &raw)118 if errors.Is(err, sql.ErrNoRows) {119 continue120 }121 if err != nil {122 return nil, err123 }124 if cutoff > 0 && seq > cutoff {125 continue126 }127 var registration event.Event128 if json.Unmarshal([]byte(raw), ®istration) != nil {129 continue130 }131 parsed, parseErr := replication.ParsePushRegistration(registration, approved, t.RelayURL())132 if parseErr == nil {133 registrations = append(registrations, parsed)134 }135 }136 return registrations, nil137}138139func appendUniqueStrings(values []string, additions ...string) []string {140 seen := make(map[string]struct{}, len(values)+len(additions))141 for _, value := range values {142 seen[value] = struct{}{}143 }144 for _, value := range additions {145 if _, ok := seen[value]; ok {146 continue147 }148 seen[value] = struct{}{}149 values = append(values, value)150 }151 return values152}