Keep accepted publish intent recoverable

package outbox

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"time"
)

var ErrNoWork = errors.New("no eligible outbox row")

type Command struct {
	ID        string
	DeviceID  string
	Topic     string
	Payload   []byte
	Deadline  time.Time
}

type Item struct {
	CommandID string
	DeviceID  string
	Topic     string
	Payload   []byte
	Deadline  time.Time
	Attempt   int
}

type Store struct { db *sql.DB }

func NewStore(db *sql.DB) (*Store, error) {
	if db == nil { return nil, fmt.Errorf("database connection is required") }
	return &Store{db: db}, nil
}

func (s *Store) Migrate(ctx context.Context) error {
	statements := []string{
		`CREATE TABLE IF NOT EXISTS device_commands (
			command_id TEXT PRIMARY KEY, device_id TEXT NOT NULL, state TEXT NOT NULL,
			deadline TIMESTAMPTZ NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT now())`,
		`CREATE TABLE IF NOT EXISTS command_outbox (
			command_id TEXT PRIMARY KEY REFERENCES device_commands(command_id),
			topic TEXT NOT NULL, payload BYTEA NOT NULL, attempts INTEGER NOT NULL DEFAULT 0,
			available_at TIMESTAMPTZ NOT NULL DEFAULT now(), lease_until TIMESTAMPTZ,
			published_at TIMESTAMPTZ, last_reason SMALLINT)`,
	}
	for _, statement := range statements {
		if _, err := s.db.ExecContext(ctx, statement); err != nil {
			return fmt.Errorf("create outbox schema: %w", err)
		}
	}
	return nil
}

func (s *Store) Enqueue(ctx context.Context, command Command) error {
	if command.ID == "" || command.DeviceID == "" || command.Topic == "" || len(command.Payload) == 0 {
		return fmt.Errorf("command ID, device, topic, and payload are required")
	}
	if !command.Deadline.After(time.Now()) {
		return fmt.Errorf("command deadline must be in the future")
	}
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil { return fmt.Errorf("begin command transaction: %w", err) }
	defer func() { _ = tx.Rollback() }()
	if _, err := tx.ExecContext(ctx, `INSERT INTO device_commands(command_id,device_id,state,deadline)
VALUES ($1,$2,'pending',$3)`, command.ID, command.DeviceID, command.Deadline); err != nil {
		return fmt.Errorf("insert command: %w", err)
	}
	if _, err := tx.ExecContext(ctx, `INSERT INTO command_outbox(command_id,topic,payload)
VALUES ($1,$2,$3)`, command.ID, command.Topic, command.Payload); err != nil {
		return fmt.Errorf("insert outbox intent: %w", err)
	}
	if err := tx.Commit(); err != nil { return fmt.Errorf("commit command and outbox: %w", err) }
	return nil
}

func (s *Store) ClaimOne(ctx context.Context, lease time.Duration, now time.Time) (Item, error) {
	if lease <= 0 { return Item{}, fmt.Errorf("lease must be positive") }
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil { return Item{}, fmt.Errorf("begin outbox claim: %w", err) }
	defer func() { _ = tx.Rollback() }()
	var item Item
	err = tx.QueryRowContext(ctx, `SELECT o.command_id,c.device_id,o.topic,o.payload,c.deadline,o.attempts
FROM command_outbox o JOIN device_commands c USING (command_id)
WHERE o.published_at IS NULL AND o.available_at <= $1
  AND (o.lease_until IS NULL OR o.lease_until <= $1)
  AND c.deadline > $1
ORDER BY o.available_at,o.command_id
LIMIT 1 FOR UPDATE OF o SKIP LOCKED`, now).Scan(
		&item.CommandID, &item.DeviceID, &item.Topic, &item.Payload, &item.Deadline, &item.Attempt)
	if errors.Is(err, sql.ErrNoRows) { return Item{}, ErrNoWork }
	if err != nil { return Item{}, fmt.Errorf("select outbox work: %w", err) }
	item.Payload = append([]byte(nil), item.Payload...)
	if _, err := tx.ExecContext(ctx, `UPDATE command_outbox
SET lease_until=$2, attempts=attempts+1 WHERE command_id=$1`, item.CommandID, now.Add(lease)); err != nil {
		return Item{}, fmt.Errorf("lease outbox work: %w", err)
	}
	if err := tx.Commit(); err != nil { return Item{}, fmt.Errorf("commit outbox lease: %w", err) }
	item.Attempt++
	return item, nil
}

func (s *Store) MarkPublished(ctx context.Context, commandID string, reason byte, at time.Time) error {
	result, err := s.db.ExecContext(ctx, `UPDATE command_outbox SET published_at=$2,
lease_until=NULL,last_reason=$3 WHERE command_id=$1 AND published_at IS NULL`, commandID, at, reason)
	if err != nil { return fmt.Errorf("mark outbox published: %w", err) }
	rows, err := result.RowsAffected()
	if err != nil { return fmt.Errorf("inspect outbox update: %w", err) }
	if rows != 1 { return fmt.Errorf("outbox command %q was not pending", commandID) }
	return nil
}

1. Commit the command and outbox intent atomically.
2. The worker can crash after PUBACK but before the sent marker; same-ID retry may duplicate.
3. Paho queue admission does not prove broker delivery or business completion.
4. Report unknown when the device may have acted but the application response is missing.