toki
64dd8a6664
- Add SDD documentation for webhook-revision-reconcile phase - Add task breakdown for webhook-revision-reconcile - Update ROADMAP.md with latest phase progress - Update PHASE.md files for agent-shell-iop-execution and provider-change-request-integration - Update milestone details for iop-agent-run-bridge and webhook-revision-reconcile - Implement controlplane router and runtime changes - Add router tests
640 lines
18 KiB
Go
640 lines
18 KiB
Go
package controlplane
|
|
|
|
import (
|
|
"context"
|
|
"crypto/rand"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"fmt"
|
|
"net/url"
|
|
"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/gitengine"
|
|
"git.toki-labs.com/toki/gito/services/core/internal/protosocket"
|
|
"git.toki-labs.com/toki/gito/services/core/internal/storage"
|
|
"git.toki-labs.com/toki/gito/services/core/internal/worker"
|
|
)
|
|
|
|
type EventBroadcaster interface {
|
|
BroadcastEnvelope(ctx context.Context, env protosocket.Envelope) error
|
|
}
|
|
|
|
type Runtime struct {
|
|
mu sync.Mutex
|
|
watches map[string]BranchWatch
|
|
subscriptions map[string]WebhookSubscription
|
|
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"`
|
|
}
|
|
|
|
type WebhookSubscription struct {
|
|
ID string
|
|
Name string
|
|
TargetURL string
|
|
Events []string
|
|
RepoID string
|
|
Branch string
|
|
SecretRef string
|
|
CreatedAt time.Time
|
|
}
|
|
|
|
func NewRuntime(broadcaster EventBroadcaster) *Runtime {
|
|
return &Runtime{
|
|
watches: make(map[string]BranchWatch),
|
|
subscriptions: make(map[string]WebhookSubscription),
|
|
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),
|
|
subscriptions: make(map[string]WebhookSubscription),
|
|
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) RegisterWebhookSubscription(name, targetURL string, events []string, repoID, branch, secretRef string) (WebhookSubscription, error) {
|
|
name = strings.TrimSpace(name)
|
|
targetURL = strings.TrimSpace(targetURL)
|
|
repoID = strings.TrimSpace(repoID)
|
|
branch = strings.TrimSpace(branch)
|
|
secretRef = strings.TrimSpace(secretRef)
|
|
events = normalizeEventNames(events)
|
|
if targetURL == "" {
|
|
return WebhookSubscription{}, fmt.Errorf("target_url is required")
|
|
}
|
|
parsedTarget, err := url.Parse(targetURL)
|
|
if err != nil || parsedTarget.Host == "" || (parsedTarget.Scheme != "http" && parsedTarget.Scheme != "https") {
|
|
return WebhookSubscription{}, fmt.Errorf("target_url must be an http or https URL")
|
|
}
|
|
if len(events) == 0 {
|
|
return WebhookSubscription{}, fmt.Errorf("events are required")
|
|
}
|
|
if name == "" {
|
|
name = "webhook-subscription"
|
|
}
|
|
|
|
subscription := WebhookSubscription{
|
|
ID: "webhook-sub-" + newID(),
|
|
Name: name,
|
|
TargetURL: targetURL,
|
|
Events: events,
|
|
RepoID: repoID,
|
|
Branch: branch,
|
|
SecretRef: secretRef,
|
|
CreatedAt: time.Now().UTC(),
|
|
}
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
r.subscriptions[subscription.ID] = subscription
|
|
return subscription, nil
|
|
}
|
|
|
|
func (r *Runtime) ListWebhookSubscriptions() []WebhookSubscription {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
subscriptions := make([]WebhookSubscription, 0, len(r.subscriptions))
|
|
for _, subscription := range r.subscriptions {
|
|
subscriptions = append(subscriptions, subscription)
|
|
}
|
|
sort.Slice(subscriptions, func(i, j int) bool {
|
|
return subscriptions[i].ID < subscriptions[j].ID
|
|
})
|
|
return subscriptions
|
|
}
|
|
|
|
func normalizeEventNames(events []string) []string {
|
|
seen := make(map[string]struct{}, len(events))
|
|
result := make([]string, 0, len(events))
|
|
for _, event := range events {
|
|
event = strings.TrimSpace(event)
|
|
if event == "" {
|
|
continue
|
|
}
|
|
if _, ok := seen[event]; ok {
|
|
continue
|
|
}
|
|
seen[event] = struct{}{}
|
|
result = append(result, event)
|
|
}
|
|
return result
|
|
}
|
|
|
|
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
|
|
}
|
|
|
|
var dedupeKey string
|
|
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
|
|
}
|
|
}
|
|
|
|
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 {
|
|
// Rollback delivery record on broadcast failure
|
|
if r.store.ProviderDeliveries() != nil && dedupeKey != "" {
|
|
_ = r.store.ProviderDeliveries().DeleteDelivery(ctx, provider, dedupeKey)
|
|
}
|
|
return record, true, err
|
|
}
|
|
}
|
|
|
|
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)
|
|
}
|
|
}
|
|
|
|
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)
|
|
}
|
|
|
|
type ScanRevisionOptions struct {
|
|
Provider string
|
|
RepoID string
|
|
Branch string
|
|
Workdir string
|
|
Runner gitengine.CommandRunner
|
|
ObservedAt time.Time
|
|
}
|
|
|
|
func (r *Runtime) ScanBranchRevision(ctx context.Context, opts ScanRevisionOptions) (EventRecord, bool, error) {
|
|
if r.store == nil || r.store.RevisionCursors() == nil {
|
|
return EventRecord{}, false, fmt.Errorf("revision cursor store not supported")
|
|
}
|
|
|
|
repoID := strings.TrimSpace(opts.RepoID)
|
|
branch := strings.TrimSpace(opts.Branch)
|
|
workdir := strings.TrimSpace(opts.Workdir)
|
|
if repoID == "" || branch == "" || workdir == "" {
|
|
return EventRecord{}, false, fmt.Errorf("repo_id, branch, and workdir are required")
|
|
}
|
|
|
|
provider := strings.TrimSpace(opts.Provider)
|
|
if provider == "" {
|
|
provider = "forgejo"
|
|
}
|
|
|
|
runner := opts.Runner
|
|
if runner == nil {
|
|
runner = gitengine.CLI{}
|
|
}
|
|
|
|
after, err := gitengine.HeadRevision(runner, workdir)
|
|
if err != nil {
|
|
return EventRecord{}, false, fmt.Errorf("head revision: %w", err)
|
|
}
|
|
|
|
observedAt := opts.ObservedAt
|
|
if observedAt.IsZero() {
|
|
observedAt = time.Now().UTC()
|
|
}
|
|
|
|
cursor, found, err := r.store.RevisionCursors().GetRevisionCursor(ctx, repoID, branch)
|
|
if err != nil {
|
|
return EventRecord{}, false, fmt.Errorf("get revision cursor: %w", err)
|
|
}
|
|
|
|
if !found {
|
|
newCursor := core.RevisionCursor{
|
|
RepoID: repoID,
|
|
Branch: branch,
|
|
Revision: after,
|
|
ObservedAt: observedAt,
|
|
}
|
|
if err := r.store.RevisionCursors().UpsertRevisionCursor(ctx, newCursor); err != nil {
|
|
return EventRecord{}, false, fmt.Errorf("upsert revision cursor: %w", err)
|
|
}
|
|
return EventRecord{}, false, nil
|
|
}
|
|
|
|
if cursor.Revision == after {
|
|
cursor.ObservedAt = observedAt
|
|
if err := r.store.RevisionCursors().UpsertRevisionCursor(ctx, cursor); err != nil {
|
|
return EventRecord{}, false, fmt.Errorf("upsert revision cursor: %w", err)
|
|
}
|
|
return EventRecord{}, false, nil
|
|
}
|
|
|
|
before := cursor.Revision
|
|
files, err := gitengine.ChangedFilesWithStatus(runner, workdir, before, after)
|
|
if err != nil {
|
|
return EventRecord{}, false, fmt.Errorf("changed files with status: %w", err)
|
|
}
|
|
|
|
revEvent := revisionEventFromScan(repoID, branch, before, after, files, observedAt)
|
|
record, matched, err := r.HandleRevision(ctx, provider, "", revEvent)
|
|
if err != nil {
|
|
return record, matched, err
|
|
}
|
|
if !matched {
|
|
cursor := core.RevisionCursor{
|
|
RepoID: repoID,
|
|
Branch: branch,
|
|
Revision: after,
|
|
ObservedAt: observedAt,
|
|
}
|
|
if err := r.store.RevisionCursors().UpsertRevisionCursor(ctx, cursor); err != nil {
|
|
return record, false, fmt.Errorf("upsert revision cursor: %w", err)
|
|
}
|
|
}
|
|
return record, matched, nil
|
|
}
|
|
|
|
func revisionEventFromScan(repoID, branch, before, after string, files []gitengine.ChangedFile, observedAt time.Time) core.RevisionEvent {
|
|
changedFiles := make([]core.ChangedFile, 0, len(files))
|
|
for _, f := range files {
|
|
changedFiles = append(changedFiles, core.ChangedFile{
|
|
Path: f.Path,
|
|
ChangeType: f.ChangeType,
|
|
})
|
|
}
|
|
return core.RevisionEvent{
|
|
RepoID: repoID,
|
|
Branch: branch,
|
|
Before: before,
|
|
After: after,
|
|
ChangedFiles: changedFiles,
|
|
ObservedAt: observedAt,
|
|
}
|
|
}
|
|
|
|
func (r *Runtime) ScanAgentRunRevision(ctx context.Context, req worker.RevisionScanRequest) (worker.RevisionScanResult, error) {
|
|
before := strings.TrimSpace(req.BeforeRevision)
|
|
after := strings.TrimSpace(req.AfterRevision)
|
|
|
|
if before != "" && after != "" {
|
|
// Explicit pair: bypass cursor look-up and emit immediately if watched.
|
|
// This covers cursor-free first push of a watched branch after agent_run.
|
|
record, matched, err := r.scanExplicitRevisionPair(ctx, req.Provider, req.RepoID, req.Branch, before, after, req.WorkDir, req.Runner, req.ObservedAt)
|
|
if err != nil {
|
|
return worker.RevisionScanResult{}, err
|
|
}
|
|
if matched && record.ID != "" && r.broadcaster == nil {
|
|
return worker.RevisionScanResult{}, fmt.Errorf("event matched but broadcaster is nil (not published)")
|
|
}
|
|
return worker.RevisionScanResult{
|
|
Attempted: true,
|
|
Matched: matched,
|
|
EventID: record.ID,
|
|
}, nil
|
|
}
|
|
|
|
opts := ScanRevisionOptions{
|
|
Provider: req.Provider,
|
|
RepoID: req.RepoID,
|
|
Branch: req.Branch,
|
|
Workdir: req.WorkDir,
|
|
Runner: req.Runner,
|
|
ObservedAt: req.ObservedAt,
|
|
}
|
|
record, matched, err := r.ScanBranchRevision(ctx, opts)
|
|
if err != nil {
|
|
return worker.RevisionScanResult{}, err
|
|
}
|
|
|
|
if matched && record.ID != "" && r.broadcaster == nil {
|
|
return worker.RevisionScanResult{}, fmt.Errorf("event matched but broadcaster is nil (not published)")
|
|
}
|
|
|
|
return worker.RevisionScanResult{
|
|
Attempted: true,
|
|
Matched: matched,
|
|
EventID: record.ID,
|
|
}, nil
|
|
}
|
|
|
|
// scanExplicitRevisionPair emits a branch.updated event for a known before/after
|
|
// pair, regardless of cursor state. changed_files are computed from workdir using
|
|
// runner. On publish success the cursor advances to after; on failure the cursor
|
|
// is not advanced, preserving retry semantics.
|
|
func (r *Runtime) scanExplicitRevisionPair(ctx context.Context, provider, repoID, branch, before, after, workdir string, runner gitengine.CommandRunner, observedAt time.Time) (EventRecord, bool, error) {
|
|
provider = strings.TrimSpace(provider)
|
|
if provider == "" {
|
|
provider = "forgejo"
|
|
}
|
|
if strings.TrimSpace(workdir) == "" {
|
|
return EventRecord{}, false, fmt.Errorf("workdir is required for explicit revision pair scan")
|
|
}
|
|
if runner == nil {
|
|
runner = gitengine.CLI{}
|
|
}
|
|
if observedAt.IsZero() {
|
|
observedAt = time.Now().UTC()
|
|
}
|
|
|
|
changedFiles, err := gitengine.ChangedFilesWithStatus(runner, workdir, before, after)
|
|
if err != nil {
|
|
return EventRecord{}, false, fmt.Errorf("changed files for explicit revision pair: %w", err)
|
|
}
|
|
|
|
revEvent := revisionEventFromScan(repoID, branch, before, after, changedFiles, observedAt)
|
|
|
|
record, matched, err := r.HandleRevision(ctx, provider, "", revEvent)
|
|
if err != nil {
|
|
return record, matched, err
|
|
}
|
|
|
|
return record, matched, nil
|
|
}
|
|
|
|
type outboxBroadcaster struct {
|
|
store storage.OperationEventStore
|
|
}
|
|
|
|
func NewOutboxBroadcaster(store storage.OperationEventStore) EventBroadcaster {
|
|
return &outboxBroadcaster{store: store}
|
|
}
|
|
|
|
func (o *outboxBroadcaster) BroadcastEnvelope(ctx context.Context, env protosocket.Envelope) error {
|
|
if env.Channel != "event" || env.Action != "branch.updated" {
|
|
return nil
|
|
}
|
|
|
|
payloadBytes, err := json.Marshal(env.Payload)
|
|
if err != nil {
|
|
return fmt.Errorf("marshal envelope payload: %w", err)
|
|
}
|
|
|
|
var m map[string]any
|
|
if err := json.Unmarshal(payloadBytes, &m); err != nil {
|
|
return fmt.Errorf("unmarshal envelope payload map: %w", err)
|
|
}
|
|
|
|
eventID, _ := m["id"].(string)
|
|
if eventID == "" {
|
|
eventID = fmt.Sprintf("branch-updated-outbox:%d", time.Now().UnixNano())
|
|
}
|
|
|
|
repoID, _ := m["repo_id"].(string)
|
|
subject := "repo:" + repoID
|
|
|
|
opEvent := storage.OperationEvent{
|
|
OperationID: "",
|
|
Event: events.Event{
|
|
ID: eventID,
|
|
Type: events.BranchUpdated,
|
|
Subject: subject,
|
|
Payload: payloadBytes,
|
|
CreatedAt: time.Now().UTC(),
|
|
},
|
|
}
|
|
|
|
if err := o.store.AppendEvent(ctx, opEvent); err != nil {
|
|
return fmt.Errorf("append branch.updated outbox event: %w", err)
|
|
}
|
|
return nil
|
|
}
|