mitbringsl/backend/internal/store/opstore.go

// Package store provides database access helpers for the mitbringsl backend.
// Op-log operations: idempotent append + pull, with inline LWW projection.
package store

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"strings"
	"time"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"

	hlcsync "github.com/mitbringsl/backend/internal/sync"
)

// OpType constants – must stay in sync with op_log.op_type check.
const (
	OpListCreate = "list_create"
	OpListRename = "list_rename"
	OpListDelete = "list_delete"
	OpItemAdd    = "item_add"
	OpItemUpdate = "item_update"
	OpItemRemove = "item_remove"
)

// maxPushBatch is the maximum number of ops accepted in a single push request.
const maxPushBatch = 100

// pullPageSize is the maximum number of ops returned by a single pull.
const pullPageSize = 500

// IncomingOp is a single operation sent by the client.
type IncomingOp struct {
	ClientSeq int64           `json:"client_seq"`
	OpType    string          `json:"op_type"`
	TargetID  uuid.UUID       `json:"target_id"`
	HLCTS     int64           `json:"hlc_ts"`
	Payload   json.RawMessage `json:"payload"`
}

// OpResult is returned for each op after a push – maps client_seq → server seq + hlc.
type OpResult struct {
	ClientSeq int64 `json:"client_seq"`
	Seq       int64 `json:"seq"`
	HLCTS     int64 `json:"hlc_ts"`
}

// Op is a fully hydrated op_log row, returned on pull.
type Op struct {
	Seq       int64           `json:"seq"`
	ClientID  uuid.UUID       `json:"client_id"`
	OpType    string          `json:"op_type"`
	TargetID  uuid.UUID       `json:"target_id"`
	Payload   json.RawMessage `json:"payload"`
	ClientSeq int64           `json:"client_seq"`
	HLCTS     int64           `json:"hlc_ts"`
	CreatedAt time.Time       `json:"created_at"`
}

// OpStore handles op_log reads and writes.
type OpStore struct {
	pool *pgxpool.Pool
}

// NewOpStore creates an OpStore backed by the given pool.
func NewOpStore(pool *pgxpool.Pool) *OpStore { return &OpStore{pool: pool} }

// MaxPushBatch exposes the configured limit so handlers can validate early.
func MaxPushBatch() int { return maxPushBatch }

// AppendOps idempotently inserts the given ops for (listID, userID, clientID)
// into op_log, applies the LWW projection to items/lists within the same
// transaction, and returns one OpResult per input op.
//
// Ops that already exist (UNIQUE client_id+client_seq) are silently skipped;
// the previously stored seq+hlc is returned for them.
func (s *OpStore) AppendOps(
	ctx context.Context,
	listID, userID, clientID uuid.UUID,
	ops []IncomingOp,
) ([]OpResult, error) {
	results := make([]OpResult, len(ops))

	tx, err := s.pool.BeginTx(ctx, pgx.TxOptions{})
	if err != nil {
		return nil, fmt.Errorf("opstore: begin tx: %w", err)
	}
	defer func() { _ = tx.Rollback(ctx) }()

	for i, op := range ops {
		// Tick the server HLC, incorporating the client's value.
		serverHLC := hlcsync.Tick(op.HLCTS)

		// Validate op_type.
		if !isKnownOpType(op.OpType) {
			return nil, fmt.Errorf("opstore: unknown op_type %q at index %d", op.OpType, i)
		}

		payloadBytes := []byte(op.Payload)
		if len(payloadBytes) == 0 {
			payloadBytes = []byte("{}")
		}

		var seq int64
		var storedHLC int64

		err := tx.QueryRow(ctx, `
			INSERT INTO op_log (list_id, user_id, client_id, op_type, target_id, payload, client_seq, hlc_ts)
			VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
			ON CONFLICT (client_id, client_seq) DO NOTHING
			RETURNING seq, hlc_ts`,
			listID, userID, clientID,
			op.OpType, op.TargetID, payloadBytes,
			op.ClientSeq, serverHLC,
		).Scan(&seq, &storedHLC)

		if err == pgx.ErrNoRows {
			// Already stored – look up the existing row.
			if err2 := tx.QueryRow(ctx, `
				SELECT seq, hlc_ts FROM op_log
				WHERE client_id = $1 AND client_seq = $2`,
				clientID, op.ClientSeq,
			).Scan(&seq, &storedHLC); err2 != nil {
				return nil, fmt.Errorf("opstore: lookup existing op: %w", err2)
			}
			// Skip projection for already-applied ops.
			results[i] = OpResult{ClientSeq: op.ClientSeq, Seq: seq, HLCTS: storedHLC}
			continue
		}
		if err != nil {
			return nil, fmt.Errorf("opstore: insert op: %w", err)
		}

		// Apply the projection within the same transaction.
		if err := applyProjection(ctx, tx, listID, op, storedHLC); err != nil {
			return nil, fmt.Errorf("opstore: apply projection for op %d: %w", i, err)
		}

		results[i] = OpResult{ClientSeq: op.ClientSeq, Seq: seq, HLCTS: storedHLC}
	}

	if err := tx.Commit(ctx); err != nil {
		return nil, fmt.Errorf("opstore: commit: %w", err)
	}
	return results, nil
}

// PullOps returns up to pullPageSize ops for listID with seq > since.
// The caller can detect more pages by comparing len(result) == pullPageSize.
func (s *OpStore) PullOps(ctx context.Context, listID uuid.UUID, since int64) ([]Op, error) {
	rows, err := s.pool.Query(ctx, `
		SELECT seq, client_id, op_type, target_id, payload, client_seq, hlc_ts, created_at
		FROM op_log
		WHERE list_id = $1 AND seq > $2
		ORDER BY seq ASC
		LIMIT $3`,
		listID, since, pullPageSize,
	)
	if err != nil {
		return nil, fmt.Errorf("opstore: pull ops: %w", err)
	}
	defer rows.Close()

	var ops []Op
	for rows.Next() {
		var o Op
		if err := rows.Scan(
			&o.Seq, &o.ClientID, &o.OpType, &o.TargetID,
			&o.Payload, &o.ClientSeq, &o.HLCTS, &o.CreatedAt,
		); err != nil {
			return nil, fmt.Errorf("opstore: scan op row: %w", err)
		}
		ops = append(ops, o)
	}
	return ops, rows.Err()
}

// ---------------------------------------------------------------------------
// Projection (LWW register per entity, Tombstones)
// ---------------------------------------------------------------------------

// applyProjection updates the items/lists projection tables based on op.
// All writes use LWW semantics: UPDATE ... WHERE hlc_ts < incomingHLC.
// Tombstoned rows (deleted_at IS NOT NULL) are never revived.
func applyProjection(ctx context.Context, tx pgx.Tx, listID uuid.UUID, op IncomingOp, serverHLC int64) error {
	switch op.OpType {
	case OpListCreate:
		return projListCreate(ctx, tx, listID, op, serverHLC)
	case OpListRename:
		return projListRename(ctx, tx, op, serverHLC)
	case OpListDelete:
		return projListDelete(ctx, tx, op, serverHLC)
	case OpItemAdd:
		return projItemAdd(ctx, tx, listID, op, serverHLC)
	case OpItemUpdate:
		return projItemUpdate(ctx, tx, op, serverHLC)
	case OpItemRemove:
		return projItemRemove(ctx, tx, op, serverHLC)
	default:
		// Unknown op types are logged but do not fail the transaction.
		slog.Warn("applyProjection: unhandled op_type", "op_type", op.OpType)
		return nil
	}
}

// payloadField extracts a string field from a JSON payload.
func payloadField(payload json.RawMessage, field string) string {
	var m map[string]json.RawMessage
	if err := json.Unmarshal(payload, &m); err != nil {
		return ""
	}
	v, ok := m[field]
	if !ok {
		return ""
	}
	var s string
	if err := json.Unmarshal(v, &s); err != nil {
		return ""
	}
	return s
}

func payloadBool(payload json.RawMessage, field string) (bool, bool) {
	var m map[string]json.RawMessage
	if err := json.Unmarshal(payload, &m); err != nil {
		return false, false
	}
	v, ok := m[field]
	if !ok {
		return false, false
	}
	var b bool
	if err := json.Unmarshal(v, &b); err != nil {
		return false, false
	}
	return b, true
}

func projListCreate(ctx context.Context, tx pgx.Tx, listID uuid.UUID, op IncomingOp, hlc int64) error {
	name := payloadField(op.Payload, "name")
	ownerID := op.TargetID // for list_create, target_id IS the list id; owner comes from payload
	ownerStr := payloadField(op.Payload, "owner_id")
	if ownerStr != "" {
		if id, err := uuid.Parse(ownerStr); err == nil {
			ownerID = id
		}
	}
	// Use target_id as the list id so the client can refer to it.
	_, err := tx.Exec(ctx, `
		INSERT INTO lists (id, name, owner_id, hlc_ts)
		VALUES ($1, $2, $3, $4)
		ON CONFLICT (id) DO NOTHING`,
		op.TargetID, name, ownerID, hlc,
	)
	return err
}

func projListRename(ctx context.Context, tx pgx.Tx, op IncomingOp, hlc int64) error {
	name := payloadField(op.Payload, "name")
	if name == "" {
		return nil
	}
	_, err := tx.Exec(ctx, `
		UPDATE lists
		SET name = $1, hlc_ts = $2, updated_at = now()
		WHERE id = $3 AND hlc_ts < $2 AND deleted_at IS NULL`,
		name, hlc, op.TargetID,
	)
	return err
}

func projListDelete(ctx context.Context, tx pgx.Tx, op IncomingOp, hlc int64) error {
	_, err := tx.Exec(ctx, `
		UPDATE lists
		SET deleted_at = now(), hlc_ts = $1, updated_at = now()
		WHERE id = $2 AND hlc_ts < $1 AND deleted_at IS NULL`,
		hlc, op.TargetID,
	)
	return err
}

func projItemAdd(ctx context.Context, tx pgx.Tx, listID uuid.UUID, op IncomingOp, hlc int64) error {
	name := payloadField(op.Payload, "name")
	quantity := payloadField(op.Payload, "quantity")
	_, err := tx.Exec(ctx, `
		INSERT INTO items (id, list_id, name, quantity, hlc_ts)
		VALUES ($1, $2, $3, NULLIF($4,''), $5)
		ON CONFLICT (id) DO NOTHING`,
		op.TargetID, listID, name, quantity, hlc,
	)
	if err != nil {
		return err
	}
	// Update suggestions table.
	if name != "" {
		return upsertItemName(ctx, tx, strings.ToLower(name))
	}
	return nil
}

func projItemUpdate(ctx context.Context, tx pgx.Tx, op IncomingOp, hlc int64) error {
	// Build a dynamic UPDATE only touching provided payload fields.
	setClauses := []string{"hlc_ts = $1", "updated_at = now()"}
	args := []any{hlc, op.TargetID}

	if name := payloadField(op.Payload, "name"); name != "" {
		args = append(args, name)
		setClauses = append(setClauses, fmt.Sprintf("name = $%d", len(args)))
	}
	if qty := payloadField(op.Payload, "quantity"); qty != "" {
		args = append(args, qty)
		setClauses = append(setClauses, fmt.Sprintf("quantity = $%d", len(args)))
	}
	if checked, ok := payloadBool(op.Payload, "checked"); ok {
		args = append(args, checked)
		setClauses = append(setClauses, fmt.Sprintf("checked = $%d", len(args)))
		if checked {
			setClauses = append(setClauses, "checked_at = now()")
		} else {
			setClauses = append(setClauses, "checked_at = NULL")
		}
	}

	query := fmt.Sprintf(
		`UPDATE items SET %s WHERE id = $2 AND hlc_ts < $1 AND deleted_at IS NULL`,
		strings.Join(setClauses, ", "),
	)
	_, err := tx.Exec(ctx, query, args...)
	return err
}

func projItemRemove(ctx context.Context, tx pgx.Tx, op IncomingOp, hlc int64) error {
	_, err := tx.Exec(ctx, `
		UPDATE items
		SET deleted_at = now(), hlc_ts = $1, updated_at = now()
		WHERE id = $2 AND hlc_ts < $1 AND deleted_at IS NULL`,
		hlc, op.TargetID,
	)
	return err
}

// upsertItemName increments the usage count for a lowercased item name in the
// suggestions table. Called within the projection transaction.
func upsertItemName(ctx context.Context, tx pgx.Tx, name string) error {
	_, err := tx.Exec(ctx, `
		INSERT INTO item_names (name, usage_count, last_used_at)
		VALUES ($1, 1, now())
		ON CONFLICT (name) DO UPDATE
		  SET usage_count  = item_names.usage_count + 1,
		      last_used_at = now()`,
		name,
	)
	return err
}

// isKnownOpType validates the op_type string.
func isKnownOpType(t string) bool {
	switch t {
	case OpListCreate, OpListRename, OpListDelete,
		OpItemAdd, OpItemUpdate, OpItemRemove:
		return true
	}
	return false
}