package delivery import ( "context" "fmt" "path/filepath" "sync" "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) { if t.eng.dbManager == nil { return nil, errArchiveNoDataDir } dir := filepath.Dir(t.eng.dbManager.DBPath(webhookID)) path := filepath.Join( dir, fmt.Sprintf("archive-%s.db", webhookID), ) 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 } return w, nil }