rvbox/internal/server/store/acceptance.go

package store

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

	rvboxv1 "github.com/rvbox/rvbox/gen/go/rvbox/v1"
	"github.com/rvbox/rvbox/internal/domain"
	"google.golang.org/protobuf/proto"
)

var ErrDispatchMismatch = errors.New("command acceptance does not match dispatched generation or revision")

// RecordCommandAcceptance fences an acknowledgement to the exact dispatch that
// produced it. Repeated acknowledgements are harmless; a stale session cannot
// advance a newer dispatch or overwrite a terminal decision.
func (store *Store) RecordCommandAcceptance(ctx context.Context, issueUUID domain.UUID, clientID string, generation, revision uint64, accepted bool, now time.Time) (bool, error) {
	return store.RecordCommandAcceptanceWithRejection(ctx, issueUUID, clientID, generation, revision, accepted, nil, now)
}

// RecordCommandAcceptanceWithRejection is the wire-facing variant that keeps
// a structured client admission rejection on the command row for status and
// history. The legacy method above remains convenient for reconciliation and
// tests that only have the accepted bit.
func (store *Store) RecordCommandAcceptanceWithRejection(ctx context.Context, issueUUID domain.UUID, clientID string, generation, revision uint64, accepted bool, rejection *rvboxv1.ControlError, now time.Time) (bool, error) {
	if isZeroUUID([16]byte(issueUUID)) || clientID == "" || generation == 0 || revision == 0 || now.IsZero() {
		return false, ErrDispatchMismatch
	}
	if accepted && rejection != nil {
		return false, ErrDispatchMismatch
	}
	var storedRejection []byte
	var err error
	if rejection != nil {
		if len(rejection.GetMessage()) > 4096 || len(rejection.GetIssueUuid()) > 128 {
			return false, ErrDispatchMismatch
		}
		storedRejection, err = proto.MarshalOptions{Deterministic: true}.Marshal(rejection)
		if err != nil || len(storedRejection) > 16<<10 {
			return false, ErrDispatchMismatch
		}
	}
	store.writeMu.Lock()
	defer store.writeMu.Unlock()
	database, err := store.openDatabase()
	if err != nil {
		return false, err
	}
	targetLifecycle := 3
	if !accepted {
		targetLifecycle = 11
	}
	tx, err := database.BeginTx(ctx, nil)
	if err != nil {
		return false, err
	}
	defer tx.Rollback()
	var queueExpiry sql.NullInt64
	var lateAfterExpiry int
	if err := tx.QueryRowContext(ctx, `SELECT queue_expiry_time, late_after_expiry FROM commands WHERE issue_uuid = ? AND client_id = ? AND target_session_generation = ? AND revision = ?`, issueUUID[:], clientID, generation, revision).Scan(&queueExpiry, &lateAfterExpiry); errors.Is(err, sql.ErrNoRows) {
		_ = tx.Rollback()
		return false, ErrDispatchMismatch
	} else if err != nil {
		return false, err
	}
	query := `UPDATE commands SET lifecycle = ?`
	args := []any{targetLifecycle}
	if !accepted {
		query += `, terminal_time = ?`
		args = append(args, now.UTC().UnixNano())
	}
	if rejection != nil {
		query += `, rejection = ?`
		args = append(args, storedRejection)
	}
	// The command may have been dispatched just before its queue deadline and
	// its acceptance can arrive after that deadline. Preserve the eventual
	// client truth, but durably retain the contradiction for status/audit.
	query += `, late_after_expiry = CASE WHEN queue_expiry_time IS NOT NULL AND queue_expiry_time <= ? THEN 1 ELSE late_after_expiry END`
	args = append(args, now.UTC().UnixNano())
	query += ` WHERE issue_uuid = ? AND client_id = ? AND lifecycle = 2 AND target_session_generation = ? AND revision = ?`
	args = append(args, issueUUID[:], clientID, generation, revision)
	result, err := tx.ExecContext(ctx, query, args...)
	if err != nil {
		return false, err
	}
	changed, err := result.RowsAffected()
	if err != nil {
		return false, err
	}
	if changed == 1 {
		late := queueExpiry.Valid && now.UTC().UnixNano() >= queueExpiry.Int64
		if late && lateAfterExpiry == 0 {
			if err := appendLateExpiryAudit(ctx, tx, now.UTC().UnixNano(), clientID, [16]byte(issueUUID), now.UTC().UnixNano()); err != nil {
				return false, err
			}
		}
		if err := tx.Commit(); err != nil {
			return false, err
		}
		return true, nil
	}
	if err := tx.Commit(); err != nil {
		return false, err
	}
	var lifecycle uint32
	var storedGeneration, storedRevision uint64
	err = database.QueryRowContext(ctx, `SELECT lifecycle, COALESCE(target_session_generation, 0), revision FROM commands WHERE issue_uuid = ? AND client_id = ?`, issueUUID[:], clientID).Scan(&lifecycle, &storedGeneration, &storedRevision)
	if errors.Is(err, sql.ErrNoRows) {
		return false, ErrDispatchMismatch
	}
	if err != nil {
		return false, err
	}
	if storedGeneration == generation && storedRevision == revision && ((accepted && (lifecycle == 3 || lifecycle == 4)) || (!accepted && lifecycle == 11)) {
		return false, nil
	}
	return false, ErrDispatchMismatch
}