Commit business state before acknowledging important work

package persistence

import (
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	"example.com/mqtt5-learning/internal/contracts"
)

type ReportOutcome uint8

const (
	ReportSaved ReportOutcome = iota + 1
	ReportDuplicate
)

var ErrReportIDConflict = errors.New("report ID was reused with different content")

type SalesRepository struct {
	db *sql.DB
}

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

func (r *SalesRepository) Migrate(ctx context.Context) error {
	statements := []string{
`CREATE TABLE IF NOT EXISTS sales_reports (
    device_id TEXT NOT NULL,
    report_id TEXT NOT NULL,
    payload_hash BYTEA NOT NULL,
    occurred_at TIMESTAMPTZ NOT NULL,
    saved_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    PRIMARY KEY (device_id, report_id)
)`,
`CREATE TABLE IF NOT EXISTS sales_report_lines (
    device_id TEXT NOT NULL,
    report_id TEXT NOT NULL,
    line_number INTEGER NOT NULL,
    sku TEXT NOT NULL,
    quantity BIGINT NOT NULL CHECK (quantity > 0),
    amount_minor BIGINT NOT NULL CHECK (amount_minor >= 0),
    PRIMARY KEY (device_id, report_id, line_number),
    FOREIGN KEY (device_id, report_id)
        REFERENCES sales_reports (device_id, report_id) ON DELETE CASCADE
	)`,
	}
	for _, statement := range statements {
		if _, err := r.db.ExecContext(ctx, statement); err != nil {
			return fmt.Errorf("create sales schema: %w", err)
		}
	}
	return nil
}

func (r *SalesRepository) SaveReport(ctx context.Context, report contracts.SalesReport) (ReportOutcome, error) {
	if report.DeviceID == "" || report.ReportID == "" || len(report.Items) == 0 {
		return 0, fmt.Errorf("validated report identity and at least one item are required")
	}
	occurredAt, err := time.Parse(time.RFC3339, report.OccurredAt)
	if err != nil {
		return 0, fmt.Errorf("parse report time: %w", err)
	}
	encoded, err := json.Marshal(report)
	if err != nil {
		return 0, fmt.Errorf("encode report for deduplication: %w", err)
	}
	hash := sha256.Sum256(encoded)
	tx, err := r.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
	if err != nil {
		return 0, fmt.Errorf("begin report transaction: %w", err)
	}
	defer func() { _ = tx.Rollback() }() // No effect after a successful commit.

	result, err := tx.ExecContext(ctx, `
INSERT INTO sales_reports (device_id, report_id, payload_hash, occurred_at)
VALUES ($1, $2, $3, $4)
ON CONFLICT (device_id, report_id) DO NOTHING`,
		report.DeviceID, report.ReportID, hash[:], occurredAt)
	if err != nil {
		return 0, fmt.Errorf("insert report identity: %w", err)
	}
	inserted, err := result.RowsAffected()
	if err != nil {
		return 0, fmt.Errorf("inspect report insert: %w", err)
	}
	if inserted == 0 {
		var storedHash []byte
		if err := tx.QueryRowContext(ctx, `
SELECT payload_hash FROM sales_reports WHERE device_id=$1 AND report_id=$2`,
			report.DeviceID, report.ReportID).Scan(&storedHash); err != nil {
			return 0, fmt.Errorf("load duplicate report: %w", err)
		}
		if !equalBytes(storedHash, hash[:]) {
			return 0, ErrReportIDConflict
		}
		if err := tx.Commit(); err != nil {
			return 0, fmt.Errorf("commit duplicate observation: %w", err)
		}
		return ReportDuplicate, nil
	}
	for i, line := range report.Items {
		if _, err := tx.ExecContext(ctx, `
INSERT INTO sales_report_lines
    (device_id, report_id, line_number, sku, quantity, amount_minor)
VALUES ($1, $2, $3, $4, $5, $6)`,
			report.DeviceID, report.ReportID, i, line.SKU, line.Quantity, line.AmountMinor); err != nil {
			return 0, fmt.Errorf("insert sales line %d: %w", i, err)
		}
	}
	if err := tx.Commit(); err != nil {
		return 0, fmt.Errorf("commit sales report: %w", err)
	}
	return ReportSaved, nil
}

func equalBytes(left, right []byte) bool {
	if len(left) != len(right) {
		return false
	}
	var difference byte
	for i := range left {
		difference |= left[i] ^ right[i]
	}
	return difference == 0
}

package persistence

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

	"example.com/mqtt5-learning/internal/contracts"
)

var ErrCommandResultConflict = errors.New("command ID has a conflicting result")

type CommandRepository struct { db *sql.DB }

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

func (r *CommandRepository) 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,
			broker_acked BOOLEAN NOT NULL DEFAULT FALSE,
			deadline TIMESTAMPTZ,
			created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now())`,
		`CREATE TABLE IF NOT EXISTS command_results (
			command_id TEXT NOT NULL REFERENCES device_commands(command_id),
			device_id TEXT NOT NULL, message_id TEXT NOT NULL, status TEXT NOT NULL,
			detail TEXT NOT NULL DEFAULT '', occurred_at TIMESTAMPTZ NOT NULL,
			PRIMARY KEY(command_id,message_id), UNIQUE(device_id,message_id))`,
	}
	for _, statement := range statements {
		if _, err := r.db.ExecContext(ctx, statement); err != nil {
			return fmt.Errorf("create command schema: %w", err)
		}
	}
	return nil
}

func (r *CommandRepository) Create(ctx context.Context, commandID, deviceID string) error {
	if commandID == "" || deviceID == "" { return fmt.Errorf("command ID and device ID are required") }
	_, err := r.db.ExecContext(ctx, `INSERT INTO device_commands(command_id,device_id,state)
VALUES ($1,$2,'pending')`, commandID, deviceID)
	if err != nil { return fmt.Errorf("create command: %w", err) }
	return nil
}

func (r *CommandRepository) MarkBrokerAck(ctx context.Context, commandID string) error {
	result, err := r.db.ExecContext(ctx, `UPDATE device_commands SET broker_acked=TRUE,updated_at=now()
WHERE command_id=$1`, commandID)
	if err != nil { return fmt.Errorf("record broker acknowledgement: %w", err) }
	rows, err := result.RowsAffected()
	if err != nil { return fmt.Errorf("inspect command update: %w", err) }
	if rows != 1 { return fmt.Errorf("command %q not found", commandID) }
	return nil
}

func (r *CommandRepository) ApplyResult(ctx context.Context, result contracts.CommandResult) error {
	if result.CommandID == "" || result.DeviceID == "" || result.MessageID == "" {
		return fmt.Errorf("validated result identity is required")
	}
	occurredAt, err := time.Parse(time.RFC3339, result.OccurredAt)
	if err != nil { return fmt.Errorf("parse result time: %w", err) }
	tx, err := r.db.BeginTx(ctx, nil)
	if err != nil { return fmt.Errorf("begin result transaction: %w", err) }
	defer func() { _ = tx.Rollback() }()
	insert, err := tx.ExecContext(ctx, `INSERT INTO command_results
(command_id,device_id,message_id,status,detail,occurred_at)
VALUES ($1,$2,$3,$4,$5,$6) ON CONFLICT (command_id,message_id) DO NOTHING`,
		result.CommandID,result.DeviceID,result.MessageID,result.Status,result.Detail,occurredAt)
	if err != nil { return fmt.Errorf("insert command result: %w", err) }
	rows, err := insert.RowsAffected()
	if err != nil { return fmt.Errorf("inspect result insert: %w", err) }
	if rows == 0 {
		var oldDevice, oldMessage, oldStatus, oldDetail string
		if err := tx.QueryRowContext(ctx, `SELECT device_id,message_id,status,detail
FROM command_results WHERE command_id=$1 AND message_id=$2`, result.CommandID,result.MessageID).Scan(&oldDevice,&oldMessage,&oldStatus,&oldDetail); err != nil {
			return fmt.Errorf("load prior command result: %w", err)
		}
		if oldDevice != result.DeviceID || oldMessage != result.MessageID || oldStatus != result.Status || oldDetail != result.Detail {
			return ErrCommandResultConflict
		}
		return tx.Commit() // Identical replay: keep the prior terminal state.
	}
	updated, err := tx.ExecContext(ctx, `UPDATE device_commands SET state=$2,updated_at=now()
WHERE command_id=$1 AND device_id=$3 AND state IN ('pending','accepted','unknown')`,
		result.CommandID,result.Status,result.DeviceID)
	if err != nil { return fmt.Errorf("update command state: %w", err) }
	changed, err := updated.RowsAffected()
	if err != nil { return fmt.Errorf("inspect command state update: %w", err) }
	if changed != 1 { return fmt.Errorf("command missing, device mismatch, or terminal state conflict") }
	if err := tx.Commit(); err != nil { return fmt.Errorf("commit command result: %w", err) }
	return nil
}

1. Unique stable business key plus one transaction for identity/effect/outcome.
2. Commit then Ack; after crash-before-Ack, recognize duplicate and avoid repeating the effect.
3. Callback error alone does not request retry; recovery is an application policy.
4. Manual Ack is not a durable application queue and depends on session/broker behavior.