tinyrelayfile
tinyrelay/internal/daemon/push_followups.go/
internal/daemon/push_followups.go raw
1package daemon
2
3import (
4 "context"
5 "database/sql"
6 "encoding/json"
7 "errors"
8
9 "github.com/FelineStateMachine/tinyrelay/internal/event"
10 "github.com/FelineStateMachine/tinyrelay/internal/replication"
11 "github.com/FelineStateMachine/tinyrelay/internal/storage"
12)
13
14type 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}
21
22// prepareReplicationPushFollowup captures registrations that existed when the
23// source event was accepted. RegistrationCutoff is the SQLite insertion
24// sequence, rather than the registration event's authored timestamp; clients
25// can author addressable registrations with arbitrary timestamps. A discovery
26// error is represented in the payload so the shared event-followup worker can
27// 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 != nil
32 return payload, nil
33}
34
35// recoverReplicationPushFollowup discovers only registrations inserted no
36// later than the acceptance cutoff. This prevents a registration created while
37// 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 nil
41 }
42 ids, err := t.discoverReplicationPushRegistrations(ctx, payload.Event, payload.RegistrationCutoff)
43 if err != nil {
44 return err
45 }
46 payload.RegistrationIDs = appendUniqueStrings(payload.RegistrationIDs, ids...)
47 payload.Discover = false
48 return nil
49}
50
51func (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, err
55 }
56 return replication.PrepareCallbackIntents(ctx, payload.Event, registrations), nil
57}
58
59func (t *Tenant) planReplicationPushFollowup(ctx context.Context, payload replicationPushFollowupPayload) ([]storage.Intent, error) {
60 if !t.Policy().Features.Push {
61 return nil, nil
62 }
63 if err := t.recoverReplicationPushFollowup(ctx, &payload); err != nil {
64 return nil, err
65 }
66 return t.prepareReplicationPushIntents(ctx, payload)
67}
68
69func (t *Tenant) captureReplicationPushRegistrations(ctx context.Context, e event.Event) ([]string, int64, error) {
70 var cutoff int64
71 if err := t.store.DB().QueryRowContext(ctx, "SELECT COALESCE(MAX(seq),0) FROM events").Scan(&cutoff); err != nil {
72 return nil, 0, err
73 }
74 return t.discoverReplicationPushRegistrationsAt(ctx, e, cutoff)
75}
76
77func (t *Tenant) discoverReplicationPushRegistrations(ctx context.Context, e event.Event, cutoff int64) ([]string, error) {
78 if cutoff <= 0 {
79 return nil, nil
80 }
81 ids, _, err := t.discoverReplicationPushRegistrationsAt(ctx, e, cutoff)
82 return ids, err
83}
84
85func (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, err
89 }
90 defer rows.Close()
91 approved := t.callbackPolicy()
92 ids := []string{}
93 for rows.Next() {
94 var seq int64
95 var id, raw string
96 if err := rows.Scan(&seq, &id, &raw); err != nil {
97 return ids, cutoff, err
98 }
99 var registration event.Event
100 if json.Unmarshal([]byte(raw), &registration) != nil {
101 continue
102 }
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}
110
111func (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 int64
116 var raw string
117 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 continue
120 }
121 if err != nil {
122 return nil, err
123 }
124 if cutoff > 0 && seq > cutoff {
125 continue
126 }
127 var registration event.Event
128 if json.Unmarshal([]byte(raw), &registration) != nil {
129 continue
130 }
131 parsed, parseErr := replication.ParsePushRegistration(registration, approved, t.RelayURL())
132 if parseErr == nil {
133 registrations = append(registrations, parsed)
134 }
135 }
136 return registrations, nil
137}
138
139func 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 continue
147 }
148 seen[value] = struct{}{}
149 values = append(values, value)
150 }
151 return values
152}

About

ownerc01503e8035f
visibilitypublic
headd730679639bf

Maintainers

  • c01503e8035f owner
  • ba1b5beed5b5 agent hermes

Clone

  • https://012.run/npub1cq2s86qrtadf36eqq3epm38wmhdynrrcwxf84eldk3f94dkv7ttsrulrz5/tinyrelay.git

Refs