| 1 | package daemon |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "database/sql" |
| 6 | "encoding/json" |
| 7 | "errors" |
| 8 | "fmt" |
| 9 | "time" |
| 10 | |
| 11 | "github.com/FelineStateMachine/tinyrelay/internal/event" |
| 12 | "github.com/FelineStateMachine/tinyrelay/internal/storage" |
| 13 | "github.com/FelineStateMachine/tinyrelay/internal/telemetry" |
| 14 | "github.com/FelineStateMachine/tinyrelay/internal/work" |
| 15 | ) |
| 16 | |
| 17 | const ( |
| 18 | eventFollowupKind = "event-followups" |
| 19 | followupMaxAge = 15 * time.Minute |
| 20 | followupMaxAttempts = 3 |
| 21 | ) |
| 22 | |
| 23 | // eventFollowups turns accepted events into optional callback and transform |
| 24 | // intents using local state. Its planning record commits with the source |
| 25 | // event; feature handlers perform remote work using their own retry policies. |
| 26 | type eventFollowups struct { |
| 27 | store *storage.Store |
| 28 | callbacks *callbackService |
| 29 | views *customViewService |
| 30 | telemetry *telemetry.Telemetry |
| 31 | push func(context.Context, replicationPushFollowupPayload) ([]storage.Intent, error) |
| 32 | } |
| 33 | |
| 34 | // eventFollowupPayload captures the signed event and matching registrations. |
| 35 | // Discovery flags retain failed local lookups for retry. Released identifies |
| 36 | // repository state whose objects are available, and RecoverPush requests the |
| 37 | // client registration snapshot saved before Git promotion. |
| 38 | type eventFollowupPayload struct { |
| 39 | Event event.Event `json:"event"` |
| 40 | AcceptedAt int64 `json:"accepted_at"` |
| 41 | Released bool `json:"released,omitempty"` |
| 42 | Callbacks []string `json:"callbacks,omitempty"` |
| 43 | Views []customViewTarget `json:"views,omitempty"` |
| 44 | DiscoverCallbacks bool `json:"discover_callbacks,omitempty"` |
| 45 | DiscoverViews bool `json:"discover_views,omitempty"` |
| 46 | Push *replicationPushFollowupPayload `json:"push,omitempty"` |
| 47 | RecoverPush bool `json:"recover_push,omitempty"` |
| 48 | } |
| 49 | |
| 50 | func (f *eventFollowups) prepare(ctx context.Context, e event.Event, now int64, released bool) *storage.Intent { |
| 51 | return f.prepareWithPush(ctx, e, now, released, nil) |
| 52 | } |
| 53 | |
| 54 | // prepareWithPush captures eligible targets or lookup failures in an intent |
| 55 | // for the event transaction. It returns nil for ephemeral events and events |
| 56 | // with no eligible work. Repository webhook and view discovery starts when |
| 57 | // released is true. |
| 58 | func (f *eventFollowups) prepareWithPush(ctx context.Context, e event.Event, now int64, released bool, push *replicationPushFollowupPayload) *storage.Intent { |
| 59 | if event.IsEphemeral(e.Kind) { |
| 60 | return nil |
| 61 | } |
| 62 | payload := eventFollowupPayload{Event: e, AcceptedAt: now, Released: released, Push: push} |
| 63 | if released && push == nil { |
| 64 | var err error |
| 65 | payload.Push, err = f.releasedPush(ctx, e.ID) |
| 66 | payload.RecoverPush = err != nil |
| 67 | } |
| 68 | var callbackErr, viewErr error |
| 69 | if e.Kind != event.KIND_REPO_STATE || released { |
| 70 | payload.Callbacks, callbackErr = f.callbacks.CandidateIDs(ctx, e) |
| 71 | payload.Views, viewErr = f.views.CandidateTargets(ctx, e) |
| 72 | } |
| 73 | payload.DiscoverCallbacks = callbackErr != nil |
| 74 | payload.DiscoverViews = viewErr != nil |
| 75 | if len(payload.Callbacks) == 0 && len(payload.Views) == 0 && callbackErr == nil && viewErr == nil && payload.Push == nil && !payload.RecoverPush { |
| 76 | return nil |
| 77 | } |
| 78 | // Event and target fields contain no custom JSON marshalers. |
| 79 | raw, err := json.Marshal(payload) |
| 80 | if err != nil { |
| 81 | f.telemetry.Logger().Error("encode optional event follow-ups" , "error" , err) |
| 82 | return nil |
| 83 | } |
| 84 | target := "stored" |
| 85 | if released { |
| 86 | target = "released" |
| 87 | } |
| 88 | return &storage.Intent{Kind: eventFollowupKind, EventID: e.ID, Target: target, Payload: string(raw)} |
| 89 | } |
| 90 | |
| 91 | // try runs local planning after commit. A planning failure leaves the parent |
| 92 | // pending for the worker. Success commits its derived intents and parent |
| 93 | // completion together. Stable child IDs preserve existing job states on retry. |
| 94 | func (f *eventFollowups) try(ctx context.Context, intent *storage.Intent) { |
| 95 | if intent == nil { |
| 96 | return |
| 97 | } |
| 98 | job := work.Intent{ID: storage.IntentID(intent.Kind, intent.EventID, intent.Target), Kind: intent.Kind, EventID: intent.EventID, Target: intent.Target, Payload: intent.Payload} |
| 99 | if err := f.plan(ctx, job, true); err != nil { |
| 100 | f.telemetry.Logger().Debug("optional event follow-ups deferred" , "error" , err) |
| 101 | } |
| 102 | } |
| 103 | |
| 104 | // handle is the queue adapter. It makes exhausted planning failures terminal; |
| 105 | // the worker owns completion, retry scheduling and claim-token checks. |
| 106 | func (f *eventFollowups) handle(ctx context.Context, job work.Intent) error { |
| 107 | err := f.plan(ctx, job, false) |
| 108 | if err != nil && job.Attempts >= followupMaxAttempts { |
| 109 | return work.Stop(fmt.Errorf("optional follow-up planning exhausted: %w" , err)) |
| 110 | } |
| 111 | return err |
| 112 | } |
| 113 | |
| 114 | // plan rechecks the source, then persists the work each available feature can |
| 115 | // prepare. A feature error keeps the parent retryable while successful child |
| 116 | // intents retain their identity. pending allows the synchronous path to mark |
| 117 | // an unclaimed parent complete; a running worker completes its own claim. |
| 118 | func (f *eventFollowups) plan(ctx context.Context, job work.Intent, pending bool) error { |
| 119 | var payload eventFollowupPayload |
| 120 | if err := json.Unmarshal([]byte(job.Payload), &payload); err != nil || payload.Event.ID != job.EventID { |
| 121 | return work.Stop(errors.New("invalid optional event follow-up" )) |
| 122 | } |
| 123 | if time.Now().Unix()-payload.AcceptedAt > int64(followupMaxAge/time.Second) { |
| 124 | return work.Stop(errors.New("optional event follow-up expired" )) |
| 125 | } |
| 126 | active, err := f.sourceActive(ctx, payload.Event) |
| 127 | if err != nil { |
| 128 | return err |
| 129 | } |
| 130 | var intents []storage.Intent |
| 131 | var planningErr error |
| 132 | if active { |
| 133 | intents, planningErr = f.intents(ctx, payload) |
| 134 | } |
| 135 | err = f.store.WithTx(ctx, func(tx *sql.Tx) error { |
| 136 | if err := storage.AddIntents(ctx, tx, intents, time.Now().Unix()); err != nil { |
| 137 | return err |
| 138 | } |
| 139 | if pending && planningErr == nil { |
| 140 | _, err := work.CompletePendingTx(ctx, tx, job.ID, time.Now()) |
| 141 | return err |
| 142 | } |
| 143 | return nil |
| 144 | }) |
| 145 | return errors.Join(planningErr, err) |
| 146 | } |
| 147 | |
| 148 | func (f *eventFollowups) sourceActive(ctx context.Context, e event.Event) (bool, error) { |
| 149 | result, err := f.store.Query(ctx, event.Filter{IDs: []string{e.ID}}, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{All: true}, Limit: 1}) |
| 150 | return len(result.Events) != 0, err |
| 151 | } |
| 152 | |
| 153 | func (f *eventFollowups) intents(ctx context.Context, payload eventFollowupPayload) ([]storage.Intent, error) { |
| 154 | callbacks, callbackErr := f.callbackIntents(ctx, payload) |
| 155 | views, viewErr := f.viewIntents(ctx, payload) |
| 156 | intents := append(callbacks, views...) |
| 157 | var pushErr error |
| 158 | if payload.RecoverPush { |
| 159 | payload.Push, pushErr = f.releasedPush(ctx, payload.Event.ID) |
| 160 | } |
| 161 | if payload.Push != nil { |
| 162 | push, err := f.push(ctx, *payload.Push) |
| 163 | pushErr = err |
| 164 | intents = append(intents, push...) |
| 165 | } |
| 166 | return intents, errors.Join(callbackErr, viewErr, pushErr) |
| 167 | } |
| 168 | |
| 169 | // releasedPush preserves the registrations captured at client acceptance. |
| 170 | // Git promotion can happen later, after its initial planner has found the |
| 171 | // source still pending. Imports have no client push snapshot to recover. |
| 172 | func (f *eventFollowups) releasedPush(ctx context.Context, eventID string) (*replicationPushFollowupPayload, error) { |
| 173 | var raw string |
| 174 | err := f.store.DB().QueryRowContext(ctx, `SELECT payload FROM work_intents WHERE id=?` , storage.IntentID(eventFollowupKind, eventID, "stored" )).Scan(&raw) |
| 175 | if errors.Is(err, sql.ErrNoRows) { |
| 176 | return nil, nil |
| 177 | } |
| 178 | if err != nil { |
| 179 | return nil, err |
| 180 | } |
| 181 | var payload eventFollowupPayload |
| 182 | if err := json.Unmarshal([]byte(raw), &payload); err != nil { |
| 183 | return nil, err |
| 184 | } |
| 185 | return payload.Push, nil |
| 186 | } |
| 187 | |
| 188 | // callbackIntents rechecks captured registrations and retries failed discovery. |
| 189 | // Recovery requires a creation time strictly before acceptance. Both times |
| 190 | // have second precision, so uncertain same-second registrations are skipped. |
| 191 | func (f *eventFollowups) callbackIntents(ctx context.Context, payload eventFollowupPayload) ([]storage.Intent, error) { |
| 192 | if payload.DiscoverCallbacks { |
| 193 | ids, err := f.callbacks.CandidateIDs(ctx, payload.Event) |
| 194 | if err != nil { |
| 195 | return nil, err |
| 196 | } |
| 197 | for _, id := range ids { |
| 198 | candidate, err := f.callbacks.callback(ctx, id) |
| 199 | if err != nil { |
| 200 | return nil, err |
| 201 | } |
| 202 | if candidate.CreatedAt < payload.AcceptedAt { |
| 203 | payload.Callbacks = append(payload.Callbacks, id) |
| 204 | } |
| 205 | } |
| 206 | } |
| 207 | return f.callbacks.PrepareFor(ctx, payload.Event, payload.Callbacks) |
| 208 | } |
| 209 | |
| 210 | // viewIntents rechecks registration generations and source relevance. Failed |
| 211 | // discovery uses the same strict acceptance cutoff as callbackIntents. |
| 212 | func (f *eventFollowups) viewIntents(ctx context.Context, payload eventFollowupPayload) ([]storage.Intent, error) { |
| 213 | if payload.DiscoverViews { |
| 214 | targets, err := f.views.CandidateTargets(ctx, payload.Event) |
| 215 | if err != nil { |
| 216 | return nil, err |
| 217 | } |
| 218 | for _, target := range targets { |
| 219 | candidate, err := f.views.customViewByName(ctx, target.Name) |
| 220 | if err != nil { |
| 221 | return nil, err |
| 222 | } |
| 223 | if candidate.CreatedAt < payload.AcceptedAt { |
| 224 | payload.Views = append(payload.Views, target) |
| 225 | } |
| 226 | } |
| 227 | } |
| 228 | return f.views.PrepareTargets(ctx, payload.Event, payload.Views, payload.Released) |
| 229 | } |