1package daemon23import (4 "context"5 "errors"6 "fmt"7 "sort"8 "strings"9 "time"1011 "github.com/FelineStateMachine/tinyrelay/internal/event"12 gitrelay "github.com/FelineStateMachine/tinyrelay/tinygit"13 "github.com/FelineStateMachine/tinyrelay/internal/replication"14 "github.com/FelineStateMachine/tinyrelay/internal/storage"15)1617const maxGitLivePeers = 321819// maxGitLiveRoots bounds the conversation roots one live subscription follows.20const maxGitLiveRoots = 1002122type gitLiveWorker struct {23 cancel context.CancelFunc24 done chan struct{}25 started time.Time26}27type gitLiveTarget struct {28 repo gitrelay.Repository29 target string30}3132// runGitLive reconciles the current repository snapshot with live workers. A33// worker is removed as soon as its repository, relay, feature or private-peer34// authorization disappears. Rotation gives relays beyond the socket budget a35// turn on every reconciliation pass.36func (t *Tenant) runGitLive(ctx context.Context) {37 workers := make(map[string]gitLiveWorker)38 var cursor int39 defer stopGitLiveWorkers(workers)40 for {41 if !t.gitLiveEnabled() {42 stopGitLiveWorkers(workers)43 workers = make(map[string]gitLiveWorker)44 } else {45 cursor = t.reconcileGitLive(ctx, workers, cursor)46 }47 timer := time.NewTimer(5 * time.Second)48 select {49 case <-ctx.Done():50 timer.Stop()51 return52 case <-timer.C:53 }54 }55}5657func (t *Tenant) gitLiveEnabled() bool {58 p := t.Policy()59 return p.Features.Grasp && p.Features.Grasp02 && t.git != nil60}6162func (t *Tenant) reconcileGitLive(ctx context.Context, workers map[string]gitLiveWorker, cursor int) int {63 desired := t.gitLiveTargets(ctx)64 keys := make([]string, 0, len(desired))65 for key := range desired {66 keys = append(keys, key)67 }68 sort.Strings(keys)69 if len(keys) == 0 {70 stopGitLiveWorkers(workers)71 return 072 }73 if cursor >= len(keys) {74 cursor = 075 }76 allowed := gitLiveCohort(keys, cursor)77 now := time.Now()78 for key, worker := range workers {79 _, wanted := desired[key]80 _, inCohort := allowed[key]81 if !wanted || (!inCohort && now.Sub(worker.started) >= gitLiveDwell) {82 worker.cancel()83 if worker.done != nil {84 <-worker.done85 }86 delete(workers, key)87 }88 }89 allowedKeys := make([]string, 0, len(allowed))90 for key := range allowed {91 allowedKeys = append(allowedKeys, key)92 }93 sort.Strings(allowedKeys)94 for _, key := range allowedKeys {95 if len(workers) >= maxGitLivePeers {96 break97 }98 if _, ok := workers[key]; ok {99 continue100 }101 workerCtx, cancel := context.WithCancel(ctx)102 done := make(chan struct{})103 workers[key] = gitLiveWorker{cancel: cancel, done: done, started: now}104 go func(repo gitrelay.Repository, target string) {105 defer close(done)106 t.gitLivePeer(workerCtx, repo, target)107 }(desired[key].repo, desired[key].target)108 }109 if len(keys) > maxGitLivePeers {110 return (cursor + maxGitLivePeers) % len(keys)111 }112 return cursor113}114115func gitLiveCohort(keys []string, cursor int) map[string]struct{} {116 allowed := make(map[string]struct{}, maxGitLivePeers)117 if len(keys) == 0 {118 return allowed119 }120 cursor %= len(keys)121 if cursor < 0 {122 cursor += len(keys)123 }124 for i := 0; i < len(keys) && i < maxGitLivePeers; i++ {125 allowed[keys[(cursor+i)%len(keys)]] = struct{}{}126 }127 return allowed128}129130const gitLiveDwell = 2 * time.Minute131132func (t *Tenant) gitLiveTargets(ctx context.Context) map[string]gitLiveTarget {133 result := make(map[string]gitLiveTarget)134 for _, repo := range t.git.Repositories() {135 targets := append([]string(nil), repo.Relays...)136 if t.Policy().Features.Grasp03 {137 if authors, err := t.gitLiveConversationAuthors(ctx, repo); err == nil {138 offset := t.gitOutboxOffset(ctx, repo)139 authors = rotateGitAuthors(authors, offset)140 if lists, err := t.gitLocalOutboxEvents(ctx, authors); err == nil {141 targets = append(targets, gitOutboxRelays(lists, repo.Private, t.Policy().PrivatePeers)...)142 }143 }144 }145 for _, target := range targets {146 target = normalizeRelayGit(target)147 if target == "" || sameRelayGit(target, t.RelayURL(), t.publicURL) || !t.gitLiveAuthorized(repo, target) {148 continue149 }150 key := strings.Join([]string{repo.Owner, repo.Identifier, target, repo.EventID, fmt.Sprint(repo.Private)}, "\x00")151 result[key] = gitLiveTarget{repo: repo, target: target}152 }153 }154 return result155}156157func (t *Tenant) gitLiveConversationAuthors(ctx context.Context, repo gitrelay.Repository) ([]string, error) {158 items := make([]event.Event, 0, 256)159 for _, filter := range gitEventFilters(repo)[1:] {160 rows, err := t.store.Query(ctx, filter, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{All: true}, Limit: 256})161 if err != nil {162 return nil, err163 }164 items = append(items, rows.Events...)165 }166 return gitConversationAuthors(items, repo.Owner, repo.Maintainers), nil167}168169func rotateGitAuthors(authors []string, offset int) []string {170 if len(authors) == 0 {171 return nil172 }173 offset %= len(authors)174 if offset < 0 {175 offset += len(authors)176 }177 rotated := append([]string(nil), authors[offset:]...)178 return append(rotated, authors[:offset]...)179}180181func (t *Tenant) gitLiveAuthorized(repo gitrelay.Repository, target string) bool {182 if !repo.Private && !privatePolicy(t.Policy()) {183 return true184 }185 return privatePeerMatch(target, t.Policy().PrivatePeers)186}187188func stopGitLiveWorkers(workers map[string]gitLiveWorker) {189 for key, worker := range workers {190 worker.cancel()191 if worker.done != nil {192 <-worker.done193 }194 delete(workers, key)195 }196}197198// gitLivePeer performs bounded history overlap before opening subscriptions.199// One subscription follows all repository and conversation filters.200func (t *Tenant) gitLivePeer(ctx context.Context, repo gitrelay.Repository, target string) {201 backoff := time.Second202 for ctx.Err() == nil {203 historyCtx, historyCancel := context.WithTimeout(ctx, 20*time.Second)204 if err := t.gitSyncPeer(historyCtx, repo, target); err != nil && ctx.Err() == nil {205 t.app.telemetry.Logger().Debug("Git live history overlap incomplete", "target", target, "error", err)206 }207 historyCancel()208 conversation, err := t.gitConversationEvents(ctx, repo)209 if err != nil {210 conversation = nil211 }212 filters := gitLiveFiltersForRoots(repo, conversation)213 readyCtx, readyCancel := context.WithTimeout(ctx, 5*time.Second)214 transport, err := t.gitTargetTransport(readyCtx, repo, target)215 readyCancel()216 if err != nil {217 backoff = nextGitLiveBackoff(backoff)218 if !waitGitLive(ctx, backoff) {219 return220 }221 continue222 }223 started := time.Now()224 subCtx, subCancel := context.WithTimeout(ctx, 2*time.Minute)225 err = transport.SubscribeFilters(subCtx, target, filters, func(item event.Event) error {226 t.ingestGitLive(ctx, item)227 return nil228 })229 subCancel()230 if ctx.Err() != nil {231 return232 }233 if err != nil {234 if time.Since(started) >= 30*time.Second {235 backoff = time.Second236 }237 backoff = nextGitLiveBackoff(backoff)238 if !waitGitLive(ctx, backoff) {239 return240 }241 }242 }243}244245func (t *Tenant) ingestGitLive(ctx context.Context, item event.Event) {246 err := t.ingest(ctx, item, replication.OriginImport)247 if errors.Is(err, storage.ErrDuplicate) || errors.Is(err, storage.ErrReplaced) {248 return249 }250}251252func gitLiveFiltersForRoots(repo gitrelay.Repository, conversation []event.Event) []event.Filter {253 filters := gitEventFilters(repo)254 roots := make([]string, 0, len(conversation))255 for _, item := range conversation {256 if item.Kind == event.KIND_GIT_PATCH || item.Kind == event.KIND_GIT_PR || item.Kind == event.KIND_GIT_ISSUE {257 roots = append(roots, item.ID)258 }259 }260 if len(roots) == 0 {261 return filters262 }263 // One subscription carries every filter, so the request stays bounded to264 // the newest conversations. Older threads are covered by history passes.265 if len(roots) > maxGitLiveRoots {266 sorted := append([]event.Event(nil), conversation...)267 sortEventsNewestFirst(sorted)268 roots = roots[:0]269 for _, item := range sorted {270 if item.Kind == event.KIND_GIT_PATCH || item.Kind == event.KIND_GIT_PR || item.Kind == event.KIND_GIT_ISSUE {271 roots = append(roots, item.ID)272 }273 if len(roots) == maxGitLiveRoots {274 break275 }276 }277 }278 kinds := []int{event.KIND_GIT_PR_UPDATE, 1111, 1630, 1631, 1632, 1633}279 return append(filters,280 event.Filter{Kinds: kinds, Tags: map[string][]string{"e": roots}},281 event.Filter{Kinds: kinds, Tags: map[string][]string{"E": roots}},282 )283}284285func nextGitLiveBackoff(current time.Duration) time.Duration {286 if current >= 5*time.Minute {287 return 5 * time.Minute288 }289 return current * 2290}291292func waitGitLive(ctx context.Context, delay time.Duration) bool {293 timer := time.NewTimer(delay)294 defer timer.Stop()295 select {296 case <-ctx.Done():297 return false298 case <-timer.C:299 return true300 }301}302303func liveFilter(repo gitrelay.Repository) event.Filter {304 return gitEventFilters(repo)[1]305}