package controlplane import ( "context" "crypto/rand" "encoding/hex" "fmt" "sort" "strings" "sync" "time" "git.toki-labs.com/toki/gito/services/core/internal/core" "git.toki-labs.com/toki/gito/services/core/internal/events" "git.toki-labs.com/toki/gito/services/core/internal/protosocket" "git.toki-labs.com/toki/gito/services/core/internal/storage" ) type EventBroadcaster interface { BroadcastEnvelope(ctx context.Context, env protosocket.Envelope) error } type Runtime struct { mu sync.Mutex watches map[string]BranchWatch records []EventRecord broadcaster EventBroadcaster store storage.Store } type BranchWatch struct { ID string `json:"id"` RepoID string `json:"repo_id"` Branch string `json:"branch"` Provider string `json:"provider"` CreatedAt time.Time `json:"created_at"` } type EventRecord struct { ID string `json:"id"` Type string `json:"type"` Provider string `json:"provider"` DeliveryID string `json:"delivery_id,omitempty"` Duplicate bool `json:"duplicate,omitempty"` Revision core.RevisionEvent `json:"revision"` CreatedAt time.Time `json:"created_at"` } func NewRuntime(broadcaster EventBroadcaster) *Runtime { return &Runtime{ watches: make(map[string]BranchWatch), broadcaster: broadcaster, } } // NewRuntimeWithStore constructs a Runtime backed by a durable store. // When store is nil the behavior is identical to NewRuntime. func NewRuntimeWithStore(broadcaster EventBroadcaster, store storage.Store) *Runtime { return &Runtime{ watches: make(map[string]BranchWatch), broadcaster: broadcaster, store: store, } } func (r *Runtime) RegisterBranchWatch(repoID, branch, provider string) (BranchWatch, error) { repoID = strings.TrimSpace(repoID) branch = strings.TrimSpace(branch) provider = strings.TrimSpace(provider) if repoID == "" || branch == "" { return BranchWatch{}, fmt.Errorf("repo_id and branch are required") } if provider == "" { provider = "forgejo" } if r.store != nil && r.store.BranchWatches() != nil { cw := core.BranchWatch{ ID: "watch-" + stableWatchKey(repoID, branch, provider), Provider: provider, RepoID: repoID, Branch: branch, CreatedAt: time.Now().UTC(), } stored, err := r.store.BranchWatches().UpsertBranchWatch(context.Background(), cw) if err != nil { return BranchWatch{}, fmt.Errorf("register branch watch: %w", err) } return branchWatchFromCore(stored), nil } watch := BranchWatch{ ID: "watch-" + stableWatchKey(repoID, branch, provider), RepoID: repoID, Branch: branch, Provider: provider, CreatedAt: time.Now().UTC(), } r.mu.Lock() defer r.mu.Unlock() if existing, ok := r.watches[watch.ID]; ok { return existing, nil } r.watches[watch.ID] = watch return watch, nil } func (r *Runtime) ListBranchWatches() []BranchWatch { if r.store != nil && r.store.BranchWatches() != nil { stored, err := r.store.BranchWatches().ListBranchWatches(context.Background()) if err != nil { // fall through to in-memory on transient error } else { watches := make([]BranchWatch, 0, len(stored)) for _, w := range stored { watches = append(watches, branchWatchFromCore(w)) } sort.Slice(watches, func(i, j int) bool { return watches[i].ID < watches[j].ID }) return watches } } r.mu.Lock() defer r.mu.Unlock() watches := make([]BranchWatch, 0, len(r.watches)) for _, watch := range r.watches { watches = append(watches, watch) } sort.Slice(watches, func(i, j int) bool { return watches[i].ID < watches[j].ID }) return watches } func (r *Runtime) ListEvents() []EventRecord { r.mu.Lock() defer r.mu.Unlock() records := append([]EventRecord(nil), r.records...) sort.Slice(records, func(i, j int) bool { return records[i].CreatedAt.Before(records[j].CreatedAt) }) return records } func (r *Runtime) HandleRevision(ctx context.Context, provider, deliveryID string, revision core.RevisionEvent) (EventRecord, bool, error) { provider = strings.TrimSpace(provider) if provider == "" { provider = "forgejo" } deliveryID = strings.TrimSpace(deliveryID) record := EventRecord{ ID: "event-" + newID(), Type: events.BranchUpdated, Provider: provider, DeliveryID: deliveryID, Revision: revision, CreatedAt: time.Now().UTC().Truncate(time.Microsecond), } if r.store != nil && r.store.BranchWatches() != nil { return r.handleRevisionWithStore(ctx, provider, deliveryID, revision, record) } r.mu.Lock() matched := r.matchesLocked(provider, revision) if matched { r.records = append(r.records, record) } r.mu.Unlock() if !matched { return record, false, nil } if r.broadcaster != nil { if err := r.broadcaster.BroadcastEnvelope(ctx, branchUpdatedEnvelope(record)); err != nil { return record, true, err } } return record, true, nil } func (r *Runtime) handleRevisionWithStore(ctx context.Context, provider, deliveryID string, revision core.RevisionEvent, record EventRecord) (EventRecord, bool, error) { _, matched, err := r.store.BranchWatches().FindBranchWatch(ctx, provider, revision.RepoID, revision.Branch) if err != nil { return record, false, fmt.Errorf("find branch watch: %w", err) } if !matched { return record, false, nil } if r.store.ProviderDeliveries() != nil { dedupeKey := dedupeKeyFor(deliveryID, revision) delivery := core.ProviderDelivery{ ID: newID(), Provider: provider, DeliveryID: deliveryID, DedupeKey: dedupeKey, EventID: record.ID, RepoID: revision.RepoID, Branch: revision.Branch, Revision: revision.After, CreatedAt: record.CreatedAt, } result, err := r.store.ProviderDeliveries().RecordOnce(ctx, delivery) if err != nil { return record, true, fmt.Errorf("record delivery: %w", err) } if !result.First { // Return idempotent response using original delivery's event id and timestamp. record.ID = result.ExistingEventID record.CreatedAt = result.ExistingCreatedAt record.Duplicate = true return record, true, nil } } if r.store.RevisionCursors() != nil { cursor := core.RevisionCursor{ RepoID: revision.RepoID, Branch: revision.Branch, Revision: revision.After, ObservedAt: revision.ObservedAt, } if err := r.store.RevisionCursors().UpsertRevisionCursor(ctx, cursor); err != nil { return record, true, fmt.Errorf("upsert revision cursor: %w", err) } } r.mu.Lock() r.records = append(r.records, record) r.mu.Unlock() if r.broadcaster != nil { if err := r.broadcaster.BroadcastEnvelope(ctx, branchUpdatedEnvelope(record)); err != nil { return record, true, err } } return record, true, nil } func (r *Runtime) matchesLocked(provider string, revision core.RevisionEvent) bool { for _, watch := range r.watches { if watch.Provider == provider && watch.RepoID == revision.RepoID && watch.Branch == revision.Branch { return true } } return false } func dedupeKeyFor(deliveryID string, revision core.RevisionEvent) string { if deliveryID != "" { return "delivery:" + deliveryID } return fmt.Sprintf("revision:%s:%s:%s:%s", revision.RepoID, revision.Branch, revision.Before, revision.After) } func branchWatchFromCore(w core.BranchWatch) BranchWatch { return BranchWatch{ ID: w.ID, RepoID: w.RepoID, Branch: w.Branch, Provider: w.Provider, CreatedAt: w.CreatedAt, } } func branchUpdatedEnvelope(record EventRecord) protosocket.Envelope { changedFiles := make([]any, 0, len(record.Revision.ChangedFiles)) for _, file := range record.Revision.ChangedFiles { changedFiles = append(changedFiles, map[string]any{ "path": file.Path, "change_type": file.ChangeType, }) } return protosocket.NewEventEnvelope(events.BranchUpdated, map[string]any{ "id": record.ID, "type": record.Type, "provider": record.Provider, "delivery_id": record.DeliveryID, "repo_id": record.Revision.RepoID, "branch": record.Revision.Branch, "before": record.Revision.Before, "after": record.Revision.After, "changed_files": changedFiles, "observed_at": record.Revision.ObservedAt.UTC().Format(time.RFC3339Nano), "created_at": record.CreatedAt.UTC().Format(time.RFC3339Nano), }) } func stableWatchKey(repoID, branch, provider string) string { replacer := strings.NewReplacer("/", "-", " ", "-", "_", "-", ".", "-") value := strings.ToLower(provider + "-" + repoID + "-" + branch) value = replacer.Replace(value) value = strings.Trim(value, "-") if value == "" { return newID() } return value } func newID() string { b := make([]byte, 8) if _, err := rand.Read(b); err != nil { return fmt.Sprintf("%d", time.Now().UnixNano()) } return hex.EncodeToString(b) }