go-sip/internal/store/cleanup_control.go

package store

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
)

const cleanupBatchSize = 100

// CleanupControlReply leaves durable task state untouched. Rejection evidence
// has its own file/receipt cleanup; a successful stop first drains cancelled rows.
func (s *Store) CleanupControlReply(ctx context.Context, dispatcherID, eventID string) (CleanupResult, error) {
	event, ready, err := s.cleanupControlIdentity(dispatcherID, eventID)
	if err != nil {
		return CleanupResult{}, err
	}
	if !ready {
		return CleanupResult{RetainedReason: "control_reply_not_confirmed"}, nil
	}
	if event.Payload.Status != "applied" {
		return CleanupResult{RetainedReason: "rejection_files_require_cleanup"}, nil
	}
	if event.Payload.Action == "stop" {
		return s.CleanupStoppedTask(ctx, dispatcherID, eventID)
	}
	s.cleanupMu.Lock()
	defer s.cleanupMu.Unlock()
	tx, err := s.db.Begin()
	if err != nil {
		return CleanupResult{}, err
	}
	defer tx.Rollback()
	current, body, ready, err := confirmedCleanupFeedback(tx, dispatcherID, "task.control", eventID)
	if err != nil {
		return CleanupResult{}, err
	}
	if !ready {
		return CleanupResult{RetainedReason: "control_reply_not_confirmed"}, nil
	}
	if current.TenantID != event.TenantID || current.Payload.TaskID != event.Payload.TaskID || current.Payload.Action != event.Payload.Action || current.Payload.Status != event.Payload.Status || current.Payload.TaskState != event.Payload.TaskState {
		return CleanupResult{}, errors.New("control reply changed during cleanup")
	}
	deleted, err := tx.Exec(`DELETE FROM dispatcher_outbox WHERE dispatcher_id=? AND event_type='task.control' AND event_id=? AND confirmed=1`, dispatcherID, eventID)
	if err != nil {
		return CleanupResult{}, err
	}
	n, err := deleted.RowsAffected()
	if err != nil {
		return CleanupResult{}, err
	}
	if err := tx.Commit(); err != nil {
		return CleanupResult{}, err
	}
	return CleanupResult{DeletedRows: int(n), DataBytes: int64(len(body))}, nil
}

// CleanupStoppedTask deletes each cancelled call in its own transaction. The
// stop receipt remains durable across interrupted batches and individual faults.
func (s *Store) CleanupStoppedTask(ctx context.Context, dispatcherID, stopEventID string) (total CleanupResult, cleanupErr error) {
	if err := ctx.Err(); err != nil {
		return total, err
	}
	event, ready, err := s.cleanupControlIdentity(dispatcherID, stopEventID)
	if err != nil {
		return total, err
	}
	if !ready {
		total.RetainedReason = "stop_not_confirmed"
		return total, nil
	}
	if event.Payload.Action != "stop" || event.Payload.Status != "applied" || event.Payload.TaskState != "stopped" {
		return total, errors.New("cancelled-call cleanup requires an applied stop receipt")
	}
	var failures []error
	cursor := ""
	for {
		if err := ctx.Err(); err != nil {
			return total, errors.Join(append(failures, err)...)
		}
		rows, err := s.db.Query(`SELECT event_id FROM dispatcher_inbox WHERE dispatcher_id=? AND tenant_id=? AND task_id=? AND status='suppressed' AND event_id>? ORDER BY event_id LIMIT ?`, dispatcherID, event.TenantID, event.Payload.TaskID, cursor, cleanupBatchSize)
		if err != nil {
			return total, errors.Join(append(failures, err)...)
		}
		var ids []string
		for rows.Next() {
			var id string
			if err := rows.Scan(&id); err != nil {
				rows.Close()
				return total, errors.Join(append(failures, err)...)
			}
			ids = append(ids, id)
		}
		err = rows.Err()
		rows.Close()
		if err != nil {
			return total, errors.Join(append(failures, err)...)
		}
		for _, id := range ids {
			if err := ctx.Err(); err != nil {
				return total, errors.Join(append(failures, err)...)
			}
			result, err := s.CleanupCall(dispatcherID, id)
			total.DeletedRows += result.DeletedRows
			total.DataBytes += result.DataBytes
			if err != nil {
				slog.Error("Cancelled call cleanup failed; stop proof retained", "dispatcher_id", dispatcherID, "event_id", id, "task_id", event.Payload.TaskID, "error", err)
				failures = append(failures, fmt.Errorf("cancelled call %q: %w", id, err))
			}
		}
		if len(ids) < cleanupBatchSize {
			break
		}
		cursor = ids[len(ids)-1]
	}
	s.cleanupMu.Lock()
	defer s.cleanupMu.Unlock()
	tx, err := s.db.Begin()
	if err != nil {
		return total, errors.Join(append(failures, err)...)
	}
	defer tx.Rollback()
	current, body, ready, err := confirmedCleanupFeedback(tx, dispatcherID, "task.control", stopEventID)
	if err != nil {
		return total, errors.Join(append(failures, err)...)
	}
	if !ready {
		total.RetainedReason = "stop_not_confirmed"
		return total, errors.Join(failures...)
	}
	if current.TenantID != event.TenantID || current.Payload.TaskID != event.Payload.TaskID || current.Payload.Action != "stop" || current.Payload.Status != "applied" || current.Payload.TaskState != "stopped" {
		return total, errors.Join(append(failures, errors.New("stop receipt changed during cancelled-call cleanup"))...)
	}
	var stopped, remaining int
	if err := tx.QueryRow(`SELECT COUNT(*) FROM dispatcher_tasks WHERE dispatcher_id=? AND tenant_id=? AND task_id=? AND control_state='stopped'`, dispatcherID, event.TenantID, event.Payload.TaskID).Scan(&stopped); err != nil {
		return total, errors.Join(append(failures, err)...)
	}
	if err := tx.QueryRow(`SELECT COUNT(*) FROM dispatcher_inbox WHERE dispatcher_id=? AND tenant_id=? AND task_id=? AND status IN ('suppressed','pending')`, dispatcherID, event.TenantID, event.Payload.TaskID).Scan(&remaining); err != nil {
		return total, errors.Join(append(failures, err)...)
	}
	if stopped != 1 || remaining != 0 {
		total.RetainedReason = "cancelled_calls_not_fully_cleaned"
		return total, errors.Join(failures...)
	}
	deleted, err := tx.Exec(`DELETE FROM dispatcher_outbox WHERE dispatcher_id=? AND event_type='task.control' AND event_id=? AND confirmed=1`, dispatcherID, stopEventID)
	if err != nil {
		return total, errors.Join(append(failures, err)...)
	}
	n, err := deleted.RowsAffected()
	if err != nil {
		return total, errors.Join(append(failures, err)...)
	}
	if err := tx.Commit(); err != nil {
		return total, errors.Join(append(failures, err)...)
	}
	total.DeletedRows += int(n)
	total.DataBytes += int64(len(body))
	return total, errors.Join(failures...)
}

func (s *Store) cleanupControlIdentity(dispatcherID, eventID string) (cleanupFeedback, bool, error) {
	s.cleanupMu.Lock()
	defer s.cleanupMu.Unlock()
	tx, err := s.db.Begin()
	if err != nil {
		return cleanupFeedback{}, false, err
	}
	defer tx.Rollback()
	event, _, ready, err := confirmedCleanupFeedback(tx, dispatcherID, "task.control", eventID)
	return event, ready, err
}