package delivery import ( "context" "fmt" "path/filepath" "sync" "time" "gorm.io/gorm" "sneak.berlin/go/webhooker/internal/database" ) // databaseTarget is a no-retry target that archives the // full inbound event into a per-webhook archive SQLite file, // separate from the per-webhook event database. The event is // already persisted in the per-webhook event DB by the time // delivery runs; the database target additionally writes a // durable long-term copy into archive-{webhookID}.db and then // records a single attempt whose outcome reflects whether the // archive write succeeded. See archiveWriter for the // close/reopen, auto-recreate, and expiry semantics. type databaseTarget struct { eng *Engine mu sync.Mutex writers map[string]*archiveWriter } // Deliver implements Target. It archives the event, then // records one successful attempt and marks the delivery // delivered. An archiving error fails the delivery: the // attempt is recorded as failed with the error and the // delivery is marked failed, so a target that could not do // its one job (archiving) never reports success. The target // does not retry; the event remains durably stored in the // per-webhook event database. func (t *databaseTarget) Deliver( _ context.Context, webhookDB *gorm.DB, d *database.Delivery, _ *Task, _ Scheduler, ) { err := t.archive(d) if err != nil { t.eng.log.Error( "failed to archive event to database target", "delivery_id", d.ID, "event_id", d.EventID, "error", err, ) t.eng.recordResult( webhookDB, d, 1, false, 0, "", err.Error(), 0, ) t.eng.updateDeliveryStatus( webhookDB, d, database.DeliveryStatusFailed, ) return } t.eng.recordResult( webhookDB, d, 1, true, 0, "", "", 0, ) t.eng.updateDeliveryStatus( webhookDB, d, database.DeliveryStatusDelivered, ) } // archive writes the full event as a row into the webhook's // archive database, honouring the optional per-target expiry // parsed from the target config JSON. func (t *databaseTarget) archive(d *database.Delivery) error { webhookID := d.Event.WebhookID if webhookID == "" { return errArchiveMissingWebhookID } expiry, err := parseArchiveExpiry(d.Target.Config) if err != nil { return err } w, err := t.writerFor(webhookID) if err != nil { return err } row := archivedEvent{ EventID: d.Event.ID, WebhookID: webhookID, EntrypointID: d.Event.EntrypointID, Method: d.Event.Method, Headers: d.Event.Headers, Body: d.Event.Body, ContentType: d.Event.ContentType, } return w.write(row, expiry) } // writerFor returns the archiveWriter for a webhook, creating // and caching it on first use. Each webhook has one writer so // its close/reopen debounce state is shared across concurrent // deliveries. The archive file lives beside the per-webhook // event database in the data directory. func (t *databaseTarget) writerFor( webhookID string, ) (*archiveWriter, error) { path, err := t.archivePath(webhookID) if err != nil { return nil, err } t.mu.Lock() defer t.mu.Unlock() if t.writers == nil { t.writers = make(map[string]*archiveWriter) } w, ok := t.writers[webhookID] if !ok { w = newArchiveWriter(path, t.eng.log) t.writers[webhookID] = w } // A delivery claims the entry: even if the idle sweep created // it moments ago, it now belongs to the registry proper and // the sweep must leave it in place when it finishes. w.sweepOwned = false return w, nil } // sweepWriterFor returns the archive writer the idle sweep should // prune a webhook through, together with whether the sweep itself // created the registry entry. // // The sweep must route its prune through the registered writer so // the writer's mutex orders it against concurrent writes, but it // must never leave a registry entry behind: a sweep that ran // concurrently with the webhook's deletion would otherwise // re-create an entry that nothing will ever evict again, which is // exactly the leak eviction exists to prevent. An entry the sweep // creates is therefore marked sweep-owned and handed back to // releaseSweepWriter when the sweep is done. func (t *databaseTarget) sweepWriterFor( webhookID string, ) (*archiveWriter, bool, error) { path, err := t.archivePath(webhookID) if err != nil { return nil, false, err } t.mu.Lock() defer t.mu.Unlock() if t.writers == nil { t.writers = make(map[string]*archiveWriter) } w, ok := t.writers[webhookID] if ok { return w, false, nil } w = newArchiveWriter(path, t.eng.log) w.sweepOwned = true t.writers[webhookID] = w return w, true, nil } // releaseSweepWriter drops a registry entry that the idle sweep // created, so a sweep leaves the registry exactly as it found it. // // The entry is removed only if it is still the very writer the // sweep installed and no delivery has claimed it in the meantime // (writerFor clears sweepOwned when it hands a writer to the // write path). Both conditions are evaluated under the registry // lock, so an eviction that raced the sweep — which removes the // entry outright — simply finds nothing left to do here, and a // delivery that adopted the writer keeps a registered, evictable // one. func (t *databaseTarget) releaseSweepWriter( webhookID string, w *archiveWriter, ) { t.mu.Lock() defer t.mu.Unlock() cur, ok := t.writers[webhookID] if !ok || cur != w || !cur.sweepOwned { return } delete(t.writers, webhookID) } // archivePath returns the archive file path for a webhook: it // lives beside the per-webhook event database in the data // directory. It does not touch the filesystem. func (t *databaseTarget) archivePath( webhookID string, ) (string, error) { if t.eng.dbManager == nil { return "", errArchiveNoDataDir } dir := filepath.Dir(t.eng.dbManager.DBPath(webhookID)) return filepath.Join( dir, fmt.Sprintf("archive-%s.db", webhookID), ), nil } // evict drops a webhook's archive writer from the registry and // closes its handle, so a deleted webhook does not leave a // writer (and an open archive handle within its debounce // window) alive for the process lifetime. // // The map entry is removed under the registry lock, which is // then released before the handle is closed under the writer's // own lock: that ordering keeps the registry available to other // webhooks while an in-flight write on this one drains, and // closing under the writer's lock means eviction can never race // a write. // // Eviction is idempotent and silent for a webhook with no // writer, which is the common case: a webhook with no database // target never creates one. It never deletes the archive file. func (t *databaseTarget) evict(webhookID string) { t.mu.Lock() w, ok := t.writers[webhookID] if ok { delete(t.writers, webhookID) } t.mu.Unlock() if !ok { return } w.evict() t.eng.log.Info( "evicted archive writer", "webhook_id", webhookID, "path", w.path, ) } // sweepWebhook prunes one webhook's archive of rows older than // expiry, without requiring a write. It returns nil (nothing to // do) when the archive file does not exist, so a sweep never // creates an archive for a webhook that has a database target // but has never received an event. // // It also never leaves a registry entry behind: an entry it had // to create to reach the writer's mutex is released again once // the prune is done, so a sweep racing a webhook deletion cannot // resurrect the writer the eviction just dropped. func (t *databaseTarget) sweepWebhook( webhookID string, expiry time.Duration, ) error { path, err := t.archivePath(webhookID) if err != nil { return err } // Check before taking a writer at all: a webhook whose // archive has never been created gets no writer, no handle, // and no file. if !fileExists(path) { return nil } w, created, err := t.sweepWriterFor(webhookID) if err != nil { return err } if created { defer t.releaseSweepWriter(webhookID, w) } return w.sweepExpired(expiry) }