// Package postgres provides PostgreSQL-based implementations of port interfaces. package postgres import ( "context" "database/sql" "encoding/json" "errors" "fmt" "time" "github.com/orchard9/rdev/internal/domain" "github.com/orchard9/rdev/internal/port" ) // WebhookRepository implements port.WebhookRepository using PostgreSQL. type WebhookRepository struct { db *sql.DB } // NewWebhookRepository creates a new PostgreSQL webhook repository. func NewWebhookRepository(db *sql.DB) *WebhookRepository { return &WebhookRepository{db: db} } // Ensure WebhookRepository implements port.WebhookRepository at compile time. var _ port.WebhookRepository = (*WebhookRepository)(nil) // Create creates a new webhook subscription. func (r *WebhookRepository) Create(ctx context.Context, webhook *domain.Webhook) error { eventsJSON, err := json.Marshal(webhook.Events) if err != nil { return fmt.Errorf("marshal events: %w", err) } err = r.db.QueryRowContext(ctx, ` INSERT INTO webhooks (id, project_id, url, secret, events, enabled, created_at, updated_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $7) RETURNING created_at, updated_at `, webhook.ID, webhook.ProjectID, webhook.URL, nullString(webhook.Secret), string(eventsJSON), webhook.Enabled, time.Now()).Scan(&webhook.CreatedAt, &webhook.UpdatedAt) if err != nil { return fmt.Errorf("create webhook: %w", err) } return nil } // Update updates an existing webhook. func (r *WebhookRepository) Update(ctx context.Context, webhook *domain.Webhook) error { eventsJSON, err := json.Marshal(webhook.Events) if err != nil { return fmt.Errorf("marshal events: %w", err) } now := time.Now() result, err := r.db.ExecContext(ctx, ` UPDATE webhooks SET url = $1, secret = $2, events = $3, enabled = $4, updated_at = $5 WHERE id = $6 `, webhook.URL, nullString(webhook.Secret), string(eventsJSON), webhook.Enabled, now, webhook.ID) if err != nil { return fmt.Errorf("update webhook: %w", err) } rows, err := result.RowsAffected() if err != nil { return fmt.Errorf("rows affected: %w", err) } if rows == 0 { return domain.ErrWebhookNotFound } webhook.UpdatedAt = now return nil } // Delete deletes a webhook by ID. func (r *WebhookRepository) Delete(ctx context.Context, id domain.WebhookID) error { result, err := r.db.ExecContext(ctx, `DELETE FROM webhooks WHERE id = $1`, id) if err != nil { return fmt.Errorf("delete webhook: %w", err) } rows, err := result.RowsAffected() if err != nil { return fmt.Errorf("rows affected: %w", err) } if rows == 0 { return domain.ErrWebhookNotFound } return nil } // GetByID retrieves a webhook by ID. func (r *WebhookRepository) GetByID(ctx context.Context, id domain.WebhookID) (*domain.Webhook, error) { var webhook domain.Webhook var webhookID string var secret sql.NullString var eventsJSON string err := r.db.QueryRowContext(ctx, ` SELECT id, project_id, url, secret, events, enabled, created_at, updated_at FROM webhooks WHERE id = $1 `, id).Scan( &webhookID, &webhook.ProjectID, &webhook.URL, &secret, &eventsJSON, &webhook.Enabled, &webhook.CreatedAt, &webhook.UpdatedAt, ) if errors.Is(err, sql.ErrNoRows) { return nil, domain.ErrWebhookNotFound } if err != nil { return nil, fmt.Errorf("get webhook: %w", err) } webhook.ID = domain.WebhookID(webhookID) if secret.Valid { webhook.Secret = secret.String } if err := json.Unmarshal([]byte(eventsJSON), &webhook.Events); err != nil { return nil, fmt.Errorf("unmarshal events: %w", err) } return &webhook, nil } // ListByProject returns all webhooks for a project. func (r *WebhookRepository) ListByProject(ctx context.Context, projectID string) ([]*domain.Webhook, error) { rows, err := r.db.QueryContext(ctx, ` SELECT id, project_id, url, secret, events, enabled, created_at, updated_at FROM webhooks WHERE project_id = $1 ORDER BY created_at DESC `, projectID) if err != nil { return nil, fmt.Errorf("list webhooks: %w", err) } defer func() { _ = rows.Close() }() return scanWebhooks(rows) } // ListEnabledByProjectAndEvent returns enabled webhooks that subscribe to a specific event type. func (r *WebhookRepository) ListEnabledByProjectAndEvent(ctx context.Context, projectID string, eventType domain.WebhookEventType) ([]*domain.Webhook, error) { // Use JSON contains check - events column contains the event type rows, err := r.db.QueryContext(ctx, ` SELECT id, project_id, url, secret, events, enabled, created_at, updated_at FROM webhooks WHERE project_id = $1 AND enabled = true AND events::jsonb ? $2 ORDER BY created_at ASC `, projectID, string(eventType)) if err != nil { return nil, fmt.Errorf("list enabled webhooks: %w", err) } defer func() { _ = rows.Close() }() return scanWebhooks(rows) } // scanWebhooks scans rows into a slice of webhooks. func scanWebhooks(rows *sql.Rows) ([]*domain.Webhook, error) { var webhooks []*domain.Webhook for rows.Next() { var webhook domain.Webhook var webhookID string var secret sql.NullString var eventsJSON string if err := rows.Scan( &webhookID, &webhook.ProjectID, &webhook.URL, &secret, &eventsJSON, &webhook.Enabled, &webhook.CreatedAt, &webhook.UpdatedAt, ); err != nil { return nil, fmt.Errorf("scan webhook: %w", err) } webhook.ID = domain.WebhookID(webhookID) if secret.Valid { webhook.Secret = secret.String } if err := json.Unmarshal([]byte(eventsJSON), &webhook.Events); err != nil { return nil, fmt.Errorf("unmarshal events: %w", err) } webhooks = append(webhooks, &webhook) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("rows error: %w", err) } return webhooks, nil } // RecordDelivery records a webhook delivery attempt. func (r *WebhookRepository) RecordDelivery(ctx context.Context, delivery *domain.WebhookDelivery) error { _, err := r.db.ExecContext(ctx, ` INSERT INTO webhook_deliveries (id, webhook_id, event_type, payload, response_status, response_body, delivered_at, success, retry_count, error_message) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) `, delivery.ID, delivery.WebhookID, delivery.EventType, delivery.Payload, nullInt(delivery.ResponseStatus), nullString(delivery.ResponseBody), delivery.DeliveredAt, delivery.Success, delivery.RetryCount, nullString(delivery.ErrorMessage), ) if err != nil { return fmt.Errorf("record delivery: %w", err) } return nil } // GetDeliveries returns delivery history for a webhook. func (r *WebhookRepository) GetDeliveries(ctx context.Context, webhookID domain.WebhookID, filters *domain.WebhookDeliveryFilters) ([]*domain.WebhookDelivery, error) { if filters == nil { filters = domain.DefaultWebhookDeliveryFilters() } query := ` SELECT id, webhook_id, event_type, payload, response_status, response_body, delivered_at, success, retry_count, error_message FROM webhook_deliveries WHERE webhook_id = $1 ` args := []any{webhookID} argNum := 2 if filters.EventType != nil { query += fmt.Sprintf(" AND event_type = $%d", argNum) args = append(args, string(*filters.EventType)) argNum++ } if filters.Success != nil { query += fmt.Sprintf(" AND success = $%d", argNum) args = append(args, *filters.Success) argNum++ } query += " ORDER BY delivered_at DESC" query += fmt.Sprintf(" LIMIT $%d OFFSET $%d", argNum, argNum+1) args = append(args, filters.Limit, filters.Offset) rows, err := r.db.QueryContext(ctx, query, args...) if err != nil { return nil, fmt.Errorf("get deliveries: %w", err) } defer func() { _ = rows.Close() }() var deliveries []*domain.WebhookDelivery for rows.Next() { var delivery domain.WebhookDelivery var deliveryID, webhookIDStr string var eventType string var responseStatus sql.NullInt32 var responseBody, errorMessage sql.NullString if err := rows.Scan( &deliveryID, &webhookIDStr, &eventType, &delivery.Payload, &responseStatus, &responseBody, &delivery.DeliveredAt, &delivery.Success, &delivery.RetryCount, &errorMessage, ); err != nil { return nil, fmt.Errorf("scan delivery: %w", err) } delivery.ID = domain.WebhookDeliveryID(deliveryID) delivery.WebhookID = domain.WebhookID(webhookIDStr) delivery.EventType = domain.WebhookEventType(eventType) if responseStatus.Valid { delivery.ResponseStatus = int(responseStatus.Int32) } if responseBody.Valid { delivery.ResponseBody = responseBody.String } if errorMessage.Valid { delivery.ErrorMessage = errorMessage.String } deliveries = append(deliveries, &delivery) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("rows error: %w", err) } return deliveries, nil } // CleanupOldDeliveries removes delivery records older than the specified number of days. func (r *WebhookRepository) CleanupOldDeliveries(ctx context.Context, olderThanDays int) (int64, error) { result, err := r.db.ExecContext(ctx, ` DELETE FROM webhook_deliveries WHERE delivered_at < NOW() - INTERVAL '1 day' * $1 `, olderThanDays) if err != nil { return 0, fmt.Errorf("cleanup old deliveries: %w", err) } return result.RowsAffected() } // nullInt returns a sql.NullInt32 for optional int fields. func nullInt(i int) sql.NullInt32 { if i == 0 { return sql.NullInt32{} } return sql.NullInt32{Int32: int32(i), Valid: true} }