1package replication23import (4 "context"5 "fmt"6 "time"78 "github.com/FelineStateMachine/tinyrelay/internal/event"9 "github.com/FelineStateMachine/tinyrelay/internal/storage"10)1112// PullTransport reads events from target. The transport owns network clients13// and their lifecycle; PullOnce owns validation and storage.14type PullTransport interface {15 Query(ctx context.Context, target string, filter event.Filter) ([]event.Event, error)16}1718// PushTransport sends one event to target. PushAfter advances its cursor only19// after Send returns, including a relay duplicate acknowledgment.20type PushTransport interface {21 Send(ctx context.Context, target string, e event.Event) (DeliveryResult, error)22}2324// SyncStats summarizes one pull or push pass.25type SyncStats struct {26 Stored int27 Duplicates int28 Rejected int29 Sent int30 Refused int31}3233func PullOnce(ctx context.Context, store *storage.Store, transport PullTransport, target string, filter event.Filter, now int64) (SyncStats, error) {34 return PullOnceWithIngest(ctx, store, transport, target, filter, now, nil)35}3637func PullOnceWithIngest(ctx context.Context, store *storage.Store, transport PullTransport, target string, filter event.Filter, now int64, ingest func(context.Context, event.Event, Origin) error) (SyncStats, error) {38 if !safeRelayURL(target) {39 return SyncStats{}, fmt.Errorf("replication: invalid pull target %q", target)40 }41 items, err := transport.Query(ctx, target, filter)42 if err != nil {43 return SyncStats{}, fmt.Errorf("replication: query %s: %w", target, err)44 }45 var stats SyncStats46 for _, item := range items {47 if err := ctx.Err(); err != nil {48 return stats, err49 }50 if err := event.Validate(item); err != nil {51 stats.Rejected++52 continue53 }54 var saveErr error55 if ingest != nil {56 saveErr = ingest(ctx, item, OriginImport)57 } else {58 _, saveErr = store.Save(ctx, item, storage.SaveOptions{Now: now})59 }60 switch {61 case saveErr == nil:62 stats.Stored++63 case saveErr == storage.ErrDuplicate || saveErr == storage.ErrReplaced:64 stats.Duplicates++65 default:66 stats.Rejected++67 }68 }69 return stats, nil70}7172func PushAfter(ctx context.Context, store *storage.Store, transport PushTransport, target string, cursor int64, filter event.Filter, now int64) (SyncStats, int64, error) {73 if !safeRelayURL(target) {74 return SyncStats{}, cursor, fmt.Errorf("replication: invalid push target %q", target)75 }76 rows, err := store.After(ctx, cursor, filter, storage.QueryOptions{Now: now, Access: storage.Access{All: true}, Limit: 0})77 if err != nil {78 return SyncStats{}, cursor, fmt.Errorf("replication: read push cursor: %w", err)79 }80 var stats SyncStats81 last := cursor82 for _, row := range rows {83 if err := ctx.Err(); err != nil {84 return stats, last, err85 }86 result, sendErr := transport.Send(ctx, target, row.Event)87 if sendErr != nil {88 return stats, last, fmt.Errorf("replication: push event: %w", sendErr)89 }90 if result.Accepted || isDuplicate(result.Message) {91 stats.Sent++92 } else {93 stats.Refused++94 }95 last = row.Sequence96 }97 return stats, last, nil98}99100func NextRun(spec JobSpec, finished time.Time) time.Time {101 if spec.Every <= 0 {102 return time.Time{}103 }104 return finished.Add(time.Duration(spec.Every) * time.Hour)105}