1package tinygit23import (4 "context"5 "crypto/sha256"6 "encoding/hex"7 "encoding/json"8 "errors"9 "fmt"10 "os"11 "path/filepath"12 "sort"13 "strconv"14 "strings"15 "time"1617 event "github.com/FelineStateMachine/tinyrelay/protocol/nostr"18)1920// Service is the durable GRASP scheduler boundary. Tick is safe to call from21// a daemon ticker and runs one pass over the currently accepted repositories.22// Callback progress is persisted so a restart does not lose the last pass.23type Service struct {24 relay *GitRelay25}2627// ValidateImported admits metadata received through an explicitly28// synchronized relay. Local announcements continue through Validate.29func (g *GitRelay) ValidateImported(ctx context.Context, e event.Event) (Repository, error) {30 if err := event.Validate(e); err != nil {31 return Repository{}, err32 }33 if e.Kind != 30617 && e.Kind != 30618 {34 return Repository{}, fmt.Errorf("unsupported: event kind %d", e.Kind)35 }36 r, err := g.parseRepository(e)37 if err != nil {38 return Repository{}, err39 }40 if g.authorize != nil {41 if err := g.authorize(ctx, e, r); err != nil {42 return Repository{}, err43 }44 }45 if e.Kind == 30618 {46 if r.EventID == "" {47 return Repository{}, errors.New("blocked: repository announcement is missing")48 }49 if err := g.validateRefs(r.Refs); err != nil {50 return Repository{}, err51 }52 }53 return r, nil54}5556const (57 graspRetryDelay = int64(5 * 60)58 graspSuccessDelay = int64(55 * 60)59)6061type tickProgress struct {62 At int64 `json:"at"`63 LastKey string `json:"last_key"`64 Repos map[string]tickRepository `json:"repos,omitempty"`65}6667type tickRepository struct {68 AttemptedAt int64 `json:"attempted_at"`69 SucceededAt int64 `json:"succeeded_at,omitempty"`70 NextAt int64 `json:"next_at,omitempty"`71 Fingerprint string `json:"fingerprint,omitempty"`72 Attempts int `json:"attempts"`73 Error string `json:"error,omitempty"`74}7576// RepositorySyncStatus is the durable status of one GRASP reconciliation job.77type RepositorySyncStatus struct {78 AttemptedAt int64 `json:"attempted_at"`79 SucceededAt int64 `json:"succeeded_at,omitempty"`80 NextAt int64 `json:"next_at,omitempty"`81 Fingerprint string `json:"fingerprint,omitempty"`82 Attempts int `json:"attempts"`83 Error string `json:"error,omitempty"`84}8586// SyncStatus is a read-only snapshot suitable for relay status pages.87type SyncStatus struct {88 At int64 `json:"at"`89 LastKey string `json:"last_key"`90 Repositories map[string]RepositorySyncStatus `json:"repositories,omitempty"`91}9293func (g *GitRelay) GRASPService() *Service { return &Service{relay: g} }9495func (s *Service) progressPath() string {96 return filepath.Join(s.relay.root, ".tinyrelay", "grasp-progress.json")97}9899func (s *Service) SyncStatus(ctx context.Context) (SyncStatus, error) {100 if s == nil || s.relay == nil {101 return SyncStatus{}, errors.New("git relay: nil GRASP service")102 }103 select {104 case <-ctx.Done():105 return SyncStatus{}, ctx.Err()106 default:107 }108 b, err := os.ReadFile(s.progressPath())109 if errors.Is(err, os.ErrNotExist) {110 return SyncStatus{Repositories: make(map[string]RepositorySyncStatus)}, nil111 }112 if err != nil {113 return SyncStatus{}, err114 }115 var progress tickProgress116 if err := json.Unmarshal(b, &progress); err != nil {117 return SyncStatus{}, fmt.Errorf("decode GRASP progress: %w", err)118 }119 status := SyncStatus{At: progress.At, LastKey: progress.LastKey, Repositories: make(map[string]RepositorySyncStatus, len(progress.Repos))}120 for key, repo := range progress.Repos {121 status.Repositories[key] = RepositorySyncStatus{AttemptedAt: repo.AttemptedAt, SucceededAt: repo.SucceededAt, NextAt: repo.NextAt, Fingerprint: repo.Fingerprint, Attempts: repo.Attempts, Error: repo.Error}122 }123 return status, nil124}125126func (s *Service) Tick(ctx context.Context) error {127 if s == nil || s.relay == nil {128 return errors.New("git relay: nil GRASP service")129 }130 s.relay.tickMu.Lock()131 defer s.relay.tickMu.Unlock()132 s.relay.mu.RLock()133 repos := make([]Repository, 0, len(s.relay.repos))134 for _, r := range s.relay.repos {135 repos = append(repos, r)136 }137 s.relay.mu.RUnlock()138 sort.Slice(repos, func(i, j int) bool {139 return key(repos[i].Owner, repos[i].Identifier) < key(repos[j].Owner, repos[j].Identifier)140 })141 var last string142 progress := tickProgress{At: time.Now().Unix(), Repos: make(map[string]tickRepository, len(repos))}143 previousProgress, _ := s.loadProgress()144 // Resume after the last attempted repository if the prior pass ran out145 // of time. Preserve untouched statuses instead of marking them failed.146 last = previousProgress.LastKey147 for _, r := range repos {148 if previous, ok := previousProgress.Repos[key(r.Owner, r.Identifier)]; ok {149 progress.Repos[key(r.Owner, r.Identifier)] = previous150 }151 }152 start := sort.Search(len(repos), func(i int) bool { return key(repos[i].Owner, repos[i].Identifier) > last })153 repos = append(repos[start:], repos[:start]...)154 var tickErr error155 for _, r := range repos {156 if ctx.Err() != nil {157 tickErr = errors.Join(tickErr, ctx.Err())158 break159 }160 repoKey := key(r.Owner, r.Identifier)161 status := tickRepository{AttemptedAt: progress.At, Attempts: 1}162 profile := s.relay.policyValue()163 status.Fingerprint = repoFingerprint(r, profile)164 if previous, ok := previousProgress.Repos[repoKey]; ok {165 if previous.NextAt > progress.At && previous.Fingerprint == status.Fingerprint {166 progress.Repos[repoKey] = previous167 continue168 }169 if previous.Error != "" {170 status.Attempts = previous.Attempts + 1171 }172 }173 last = repoKey174 features := profile.Features175 var repoErr error176 if features.Grasp && features.Grasp02 {177 if s.relay.eventSync != nil {178 eventCtx, cancel := context.WithTimeout(ctx, 40*time.Second)179 repoErr = errors.Join(repoErr, s.relay.eventSync(eventCtx, r))180 cancel()181 }182 gitCtx, cancel := context.WithTimeout(ctx, 20*time.Second)183 if s.relay.gitSync != nil {184 repoErr = errors.Join(repoErr, s.relay.gitSync(gitCtx, r))185 } else if len(r.Refs) > 0 && len(RepairSources(r)) > 0 {186 repoErr = errors.Join(repoErr, s.relay.FetchMissing(gitCtx, r, RepairSources(r), r.Refs))187 }188 cancel()189 }190 if errors.Is(repoErr, ErrIncomplete) {191 // Progress is saved; continue soon without recording a failure.192 if previous, ok := previousProgress.Repos[repoKey]; ok {193 status.SucceededAt = previous.SucceededAt194 }195 status.NextAt = progress.At + graspRetryDelay196 } else if repoErr != nil {197 status.Error = repoErr.Error()198 status.NextAt = progress.At + graspRetryDelay199 tickErr = errors.Join(tickErr, fmt.Errorf("%s synchronization: %w", last, repoErr))200 } else {201 status.SucceededAt = progress.At202 status.NextAt = progress.At + graspSuccessDelay203 }204 progress.Repos[last] = status205 }206 progress.LastKey = last207 b, err := json.Marshal(progress)208 if err != nil {209 return err210 }211 if err := os.MkdirAll(filepath.Dir(s.progressPath()), 0700); err != nil {212 return err213 }214 tmp := s.progressPath() + ".tmp"215 if err := os.WriteFile(tmp, b, 0600); err != nil {216 return err217 }218 if err := syncFile(tmp); err != nil {219 return err220 }221 if err := os.Rename(tmp, s.progressPath()); err != nil {222 return err223 }224 return tickErr225}226227func repoFingerprint(r Repository, profile Policy) string {228 refs := make([]string, 0, len(r.Refs))229 for ref, oid := range r.Refs {230 refs = append(refs, ref+"="+oid)231 }232 sort.Strings(refs)233 profileJSON, _ := json.Marshal(struct {234 Grasp bool `json:"grasp"`235 Reads string `json:"reads"`236 Grasp02 bool `json:"grasp02"`237 Grasp03 bool `json:"grasp03"`238 Peers []string `json:"peers"`239 }{profile.Features.Grasp, profile.Reads, profile.Features.Grasp02, profile.Features.Grasp03, profile.PrivatePeers})240 parts := []string{r.Owner, r.Identifier, r.EventID, r.Head, strconv.FormatBool(r.Private), strings.Join(refs, "\x00"), strings.Join(r.Clone, "\x00"), strings.Join(r.Relays, "\x00"), strings.Join(r.Maintainers, "\x00"), string(profileJSON)}241 sum := sha256.Sum256([]byte(strings.Join(parts, "\x01")))242 return hex.EncodeToString(sum[:])243}244245func (s *Service) loadProgress() (tickProgress, error) {246 b, err := os.ReadFile(s.progressPath())247 if err != nil {248 return tickProgress{}, err249 }250 var progress tickProgress251 if err := json.Unmarshal(b, &progress); err != nil {252 return tickProgress{}, err253 }254 return progress, nil255}256257// RepairSources gives a daemon a deterministic source order for a missing258// object. Clone URLs are preferred, followed by relay URLs; callers must still259// pass each URL through AdmitSource and verify the object graph.260func RepairSources(r Repository) []string {261 out := append([]string(nil), r.Clone...)262 out = append(out, r.Relays...)263 return out264}