1package daemon23import (4 "context"5 "errors"6 "fmt"7 "time"89 "github.com/FelineStateMachine/tinyrelay/internal/event"10 gitrelay "github.com/FelineStateMachine/tinyrelay/tinygit"11 "github.com/FelineStateMachine/tinyrelay/internal/replication"12 "github.com/FelineStateMachine/tinyrelay/internal/storage"13)1415const gitOutboxOffsetPrefix = "grasp03.outbox-offset."1617type tenantGitOutboxTransport struct {18 tenant *Tenant19 repo gitrelay.Repository20}2122func (t tenantGitOutboxTransport) Query(ctx context.Context, target string, filter event.Filter) ([]event.Event, error) {23 requestCtx, cancel := context.WithTimeout(ctx, 5*time.Second)24 defer cancel()25 transport, err := t.tenant.gitTargetTransport(requestCtx, t.repo, target)26 if err != nil {27 return nil, err28 }29 return transport.Query(requestCtx, target, filter)30}3132// gitOutboxSync imports participant relay lists, then asks each discovered33// outbox for the repository conversation. List events are ingested first so34// the discovered routing remains durable after a restart.35func (t *Tenant) gitOutboxSync(ctx context.Context, repo gitrelay.Repository) error {36 if t == nil || t.store == nil {37 return errors.New("GRASP-03: tenant store is required")38 }39 events, err := t.gitConversationEvents(ctx, repo)40 if err != nil {41 return err42 }43 authors := gitConversationAuthors(events, repo.Owner, repo.Maintainers)44 offset := t.gitOutboxOffset(ctx, repo)45 selectedAuthors := boundedGitAuthorsFrom(authors, offset%maxGitInt(len(authors), 1))46 private := repo.Private || t.PrivateServiceEnabled()47 options := gitOutboxOptions{Transport: tenantGitOutboxTransport{tenant: t, repo: repo}, Seeds: repo.Relays, Private: private, PrivatePeers: t.Policy().PrivatePeers, AuthorOffset: 0, SeedOffset: offset}48 listEvents, err := t.gitLocalOutboxEvents(ctx, selectedAuthors)49 if err != nil {50 return err51 }52 discovered, discoverErr := discoverGitOutboxEvents(ctx, options, selectedAuthors)53 listEvents = mergeGitOutboxEvents(listEvents, discovered)54 var syncErrors []error55 if err := t.ingestGitOutboxEvents(ctx, discovered); err != nil {56 syncErrors = append(syncErrors, err)57 }58 if discoverErr != nil {59 syncErrors = append(syncErrors, discoverErr)60 }61 if len(authors) > 0 {62 if err := t.store.PutSetting(ctx, gitOutboxOffsetKey(repo), (offset+gitOutboxAuthorLimit)%len(authors)); err != nil {63 syncErrors = append(syncErrors, fmt.Errorf("GRASP-03: save outbox offset: %w", err))64 }65 }66 targets := gitOutboxRelays(listEvents, options.Private, options.PrivatePeers)67 targets = rotateGitRelays(targets, offset)68 for _, target := range targets {69 targetCtx, cancel := context.WithTimeout(ctx, 8*time.Second)70 if err := t.gitSyncPeer(targetCtx, repo, target); err != nil {71 syncErrors = append(syncErrors, fmt.Errorf("GRASP-03: sync %s: %w", target, err))72 }73 cancel()74 }75 return errors.Join(syncErrors...)76}7778func (t *Tenant) gitLocalOutboxEvents(ctx context.Context, authors []string) ([]event.Event, error) {79 result := make([]event.Event, 0, len(authors)*3)80 for _, author := range authors {81 for _, kind := range []int{event.KIND_PROFILE, 10002, 10317} {82 rows, err := t.store.Query(ctx, event.Filter{Authors: []string{author}, Kinds: []int{kind}, Tags: map[string][]string{}}, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{All: true}, Limit: 1})83 if err != nil {84 return nil, fmt.Errorf("GRASP-03: read cached outbox: %w", err)85 }86 if len(rows.Events) > 0 {87 result = append(result, rows.Events[0])88 }89 }90 }91 return result, nil92}9394func rotateGitRelays(relays []string, offset int) []string {95 if len(relays) == 0 {96 return nil97 }98 start := offset % len(relays)99 return append(append([]string(nil), relays[start:]...), relays[:start]...)100}101102func maxGitInt(a, b int) int {103 if a > b {104 return a105 }106 return b107}108109func mergeGitOutboxEvents(existing, incoming []event.Event) []event.Event {110 result := append([]event.Event(nil), existing...)111 for _, item := range incoming {112 found := false113 for i := range result {114 if result[i].PubKey != item.PubKey || result[i].Kind != item.Kind {115 continue116 }117 found = true118 if item.CreatedAt > result[i].CreatedAt || (item.CreatedAt == result[i].CreatedAt && item.ID < result[i].ID) {119 result[i] = item120 }121 break122 }123 if !found {124 result = append(result, item)125 }126 }127 return result128}129130func gitOutboxRelays(items []event.Event, private bool, peers []string) []string {131 result := make([]string, 0, gitOutboxRelayLimit)132 seen := make(map[string]struct{}, gitOutboxRelayLimit)133 for _, item := range items {134 for _, raw := range gitOutboxRelayTags(item) {135 if !gitOutboxAllowedRelay(raw, private, peers) {136 continue137 }138 canonical := normalizeRelayGit(raw)139 if _, ok := seen[canonical]; ok {140 continue141 }142 seen[canonical] = struct{}{}143 result = append(result, canonical)144 if len(result) == gitOutboxRelayLimit {145 return result146 }147 }148 }149 return result150}151152func (t *Tenant) ingestGitOutboxEvents(ctx context.Context, items []event.Event) error {153 for _, item := range items {154 if err := t.ingest(ctx, item, replication.OriginImport); err != nil && !errors.Is(err, storage.ErrDuplicate) && !errors.Is(err, storage.ErrReplaced) {155 return fmt.Errorf("GRASP-03: import outbox list: %w", err)156 }157 }158 return nil159}160161func (t *Tenant) gitOutboxOffset(ctx context.Context, repo gitrelay.Repository) int {162 var offset int163 if err := t.store.GetSetting(ctx, gitOutboxOffsetKey(repo), &offset); err != nil || offset < 0 {164 return 0165 }166 return offset167}168169func gitOutboxOffsetKey(repo gitrelay.Repository) string {170 return gitOutboxOffsetPrefix + repo.Owner + ":" + repo.Identifier171}