1package daemon23import (4 "context"5 "database/sql"6 "encoding/json"7 "errors"8 "fmt"9 "time"1011 "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)1617const (18 eventFollowupKind = "event-followups"19 followupMaxAge = 15 * time.Minute20 followupMaxAttempts = 321)2223// eventFollowups turns accepted events into optional callback and transform24// intents using local state. Its planning record commits with the source25// event; feature handlers perform remote work using their own retry policies.26type eventFollowups struct {27 store *storage.Store28 callbacks *callbackService29 views *customViewService30 telemetry *telemetry.Telemetry31 push func(context.Context, replicationPushFollowupPayload) ([]storage.Intent, error)32}3334// eventFollowupPayload captures the signed event and matching registrations.35// Discovery flags retain failed local lookups for retry. Released identifies36// repository state whose objects are available, and RecoverPush requests the37// client registration snapshot saved before Git promotion.38type 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}4950func (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}5354// prepareWithPush captures eligible targets or lookup failures in an intent55// for the event transaction. It returns nil for ephemeral events and events56// with no eligible work. Repository webhook and view discovery starts when57// released is true.58func (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 nil61 }62 payload := eventFollowupPayload{Event: e, AcceptedAt: now, Released: released, Push: push}63 if released && push == nil {64 var err error65 payload.Push, err = f.releasedPush(ctx, e.ID)66 payload.RecoverPush = err != nil67 }68 var callbackErr, viewErr error69 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 != nil74 payload.DiscoverViews = viewErr != nil75 if len(payload.Callbacks) == 0 && len(payload.Views) == 0 && callbackErr == nil && viewErr == nil && payload.Push == nil && !payload.RecoverPush {76 return nil77 }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 nil83 }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}9091// try runs local planning after commit. A planning failure leaves the parent92// pending for the worker. Success commits its derived intents and parent93// completion together. Stable child IDs preserve existing job states on retry.94func (f *eventFollowups) try(ctx context.Context, intent *storage.Intent) {95 if intent == nil {96 return97 }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}103104// handle is the queue adapter. It makes exhausted planning failures terminal;105// the worker owns completion, retry scheduling and claim-token checks.106func (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 err112}113114// plan rechecks the source, then persists the work each available feature can115// prepare. A feature error keeps the parent retryable while successful child116// intents retain their identity. pending allows the synchronous path to mark117// an unclaimed parent complete; a running worker completes its own claim.118func (f *eventFollowups) plan(ctx context.Context, job work.Intent, pending bool) error {119 var payload eventFollowupPayload120 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 err129 }130 var intents []storage.Intent131 var planningErr error132 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 err138 }139 if pending && planningErr == nil {140 _, err := work.CompletePendingTx(ctx, tx, job.ID, time.Now())141 return err142 }143 return nil144 })145 return errors.Join(planningErr, err)146}147148func (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, err151}152153func (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 error158 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 = err164 intents = append(intents, push...)165 }166 return intents, errors.Join(callbackErr, viewErr, pushErr)167}168169// releasedPush preserves the registrations captured at client acceptance.170// Git promotion can happen later, after its initial planner has found the171// source still pending. Imports have no client push snapshot to recover.172func (f *eventFollowups) releasedPush(ctx context.Context, eventID string) (*replicationPushFollowupPayload, error) {173 var raw string174 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, nil177 }178 if err != nil {179 return nil, err180 }181 var payload eventFollowupPayload182 if err := json.Unmarshal([]byte(raw), &payload); err != nil {183 return nil, err184 }185 return payload.Push, nil186}187188// callbackIntents rechecks captured registrations and retries failed discovery.189// Recovery requires a creation time strictly before acceptance. Both times190// have second precision, so uncertain same-second registrations are skipped.191func (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, err196 }197 for _, id := range ids {198 candidate, err := f.callbacks.callback(ctx, id)199 if err != nil {200 return nil, err201 }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}209210// viewIntents rechecks registration generations and source relevance. Failed211// discovery uses the same strict acceptance cutoff as callbackIntents.212func (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, err217 }218 for _, target := range targets {219 candidate, err := f.views.customViewByName(ctx, target.Name)220 if err != nil {221 return nil, err222 }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}