go-sip/internal/store/task_trunk_concurrency_test.go

package store

import (
	"encoding/json"
	"errors"
	"fmt"
	"path/filepath"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"git.ipao.vip/rogee/go-sip/internal/configread"
)

func trunkLimitSnapshot(t *testing.T, taskID string, revision, limit int64) configread.Snapshot {
	t.Helper()
	snapshot := currentStoreSnapshot(t)
	var task map[string]json.RawMessage
	if err := json.Unmarshal(snapshot.Task.Raw, &task); err != nil {
		t.Fatal(err)
	}
	set := func(key string, value any) {
		raw, err := json.Marshal(value)
		if err != nil {
			t.Fatal(err)
		}
		task[key] = raw
	}
	set("task_id", taskID)
	set("task_revision", revision)
	set("max_concurrent_calls", 100)
	set("allowed_trunk_ids", []configread.AllowedTrunk{{TrunkID: "trunk-mock", Concurrency: limit}})
	raw, err := json.Marshal(task)
	if err != nil {
		t.Fatal(err)
	}
	if err := json.Unmarshal(raw, &snapshot.Task); err != nil {
		t.Fatal(err)
	}
	snapshot.SIP.Trunks = []byte(strings.Replace(string(snapshot.SIP.Trunks), `"max_concurrent_calls":null`, `"max_concurrent_calls":100`, 1))
	snapshot.Quota.MaxConcurrentCalls = 100
	return snapshot
}

func taskTrunkStore(t *testing.T, snapshots ...configread.Snapshot) (*Store, string) {
	t.Helper()
	path := filepath.Join(t.TempDir(), "state.db")
	s, err := Open(path)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = s.Close() })
	tasks := make([]configread.DiscoveredTask, 0, len(snapshots))
	for _, snapshot := range snapshots {
		tasks = append(tasks, configread.DiscoveredTask{TaskID: snapshot.Task.TaskID, TenantID: snapshot.Task.TenantID, TaskRevision: snapshot.Task.TaskRevision, Status: "running"})
	}
	if err := s.ApplyDiscoverySnapshot(currentDispatcherID, tasks); err != nil {
		t.Fatal(err)
	}
	for _, snapshot := range snapshots {
		if err := s.SaveSnapshot(snapshot); err != nil {
			t.Fatal(err)
		}
	}
	if err := s.MarkReadyForSIP(currentDispatcherID, 8); err != nil {
		t.Fatal(err)
	}
	return s, path
}

func recordTrunkCall(t *testing.T, s *Store, taskID, eventID string) ExecuteCommand {
	t.Helper()
	cmd := currentCall(eventID)
	cmd.TaskID = taskID
	if _, created, err := s.RecordExecute(cmd); err != nil || !created {
		t.Fatalf("record call: created=%t err=%v", created, err)
	}
	return cmd
}

func TestTaskTrunkReservationRejectsDisabledAndUnlistedLines(t *testing.T) {
	for _, tc := range []struct {
		name, selected string
		limit          int64
	}{
		{"zero", "trunk-mock", 0},
		{"unlisted", "another-trunk", 5},
	} {
		t.Run(tc.name, func(t *testing.T) {
			s, _ := taskTrunkStore(t, trunkLimitSnapshot(t, "task-asr", 1, tc.limit))
			cmd := recordTrunkCall(t, s, "task-asr", "call-disabled")
			selection := currentReservation()
			selection.TrunkID = tc.selected
			at := time.Date(2026, 9, 21, 1, 30, 0, 0, time.UTC)
			if err := s.ReserveExecute(cmd.DispatcherID, cmd.EventID, selection, at); !errors.Is(err, ErrNotReady) {
				t.Fatalf("disabled/unlisted task line reserved: %v", err)
			}
		})
	}
}

func TestTaskTrunkLimitIsScopedToTaskAndUnknownSurvivesRestart(t *testing.T) {
	s, path := taskTrunkStore(t, trunkLimitSnapshot(t, "task-asr", 1, 1), trunkLimitSnapshot(t, "task-other", 1, 1))
	at := time.Date(2026, 9, 21, 1, 30, 0, 0, time.UTC)
	first := recordTrunkCall(t, s, "task-asr", "call-first")
	if err := s.ReserveExecute(first.DispatcherID, first.EventID, currentReservation(), at); err != nil {
		t.Fatal(err)
	}
	other := recordTrunkCall(t, s, "task-other", "call-other")
	if err := s.ReserveExecute(other.DispatcherID, other.EventID, currentReservation(), at); err != nil {
		t.Fatalf("another task lost its independent per-line capacity: %v", err)
	}
	if err := s.MarkExecuteUnknown(first.DispatcherID, first.EventID); err != nil {
		t.Fatal(err)
	}
	if err := s.Close(); err != nil {
		t.Fatal(err)
	}
	s, err := Open(path)
	if err != nil {
		t.Fatal(err)
	}
	defer s.Close()
	if err := s.MarkReadyForSIP(currentDispatcherID, 8); err != nil {
		t.Fatal(err)
	}
	occupancy, err := s.TaskTrunkOccupancy(currentDispatcherID, 1001, "task-asr")
	if err != nil || occupancy["trunk-mock"] != 1 {
		t.Fatalf("task trunk occupancy after restart=%v err=%v", occupancy, err)
	}
	second := recordTrunkCall(t, s, "task-asr", "call-second")
	if err := s.ReserveExecute(second.DispatcherID, second.EventID, currentReservation(), at); !errors.Is(err, ErrCapacity) {
		t.Fatalf("unknown call released task-line capacity on restart: %v", err)
	}
	if err := s.FinishExecute(first.DispatcherID, first.EventID); err != nil {
		t.Fatal(err)
	}
	if err := s.ReserveExecute(second.DispatcherID, second.EventID, currentReservation(), at); err != nil {
		t.Fatalf("confirmed end did not release task-line capacity: %v", err)
	}
}

func TestTaskTrunkConcurrentReservationsCannotExceedLimit(t *testing.T) {
	s, _ := taskTrunkStore(t, trunkLimitSnapshot(t, "task-asr", 1, 5))
	at := time.Date(2026, 9, 21, 1, 30, 0, 0, time.UTC)
	commands := make([]ExecuteCommand, 20)
	for i := range commands {
		commands[i] = recordTrunkCall(t, s, "task-asr", fmt.Sprintf("call-%d", i))
	}
	var accepted atomic.Int64
	var wg sync.WaitGroup
	for _, cmd := range commands {
		wg.Go(func() {
			err := s.ReserveExecute(cmd.DispatcherID, cmd.EventID, currentReservation(), at)
			if err == nil {
				accepted.Add(1)
			} else if !errors.Is(err, ErrCapacity) {
				t.Errorf("reserve: %v", err)
			}
		})
	}
	wg.Wait()
	if accepted.Load() != 5 {
		t.Fatalf("accepted=%d, want task-line limit 5", accepted.Load())
	}
}

func TestTaskTrunkEditRechecksLimitAndSelectionRevision(t *testing.T) {
	s, _ := taskTrunkStore(t, trunkLimitSnapshot(t, "task-asr", 1, 2))
	at := time.Date(2026, 9, 21, 1, 30, 0, 0, time.UTC)
	first := recordTrunkCall(t, s, "task-asr", "call-before-edit")
	if err := s.ReserveExecute(first.DispatcherID, first.EventID, currentReservation(), at); err != nil {
		t.Fatal(err)
	}
	if state, err := s.CompleteEdit(trunkLimitSnapshot(t, "task-asr", 2, 1), "edit-line-limit"); err != nil || state != "applied" {
		t.Fatalf("edit=%q err=%v", state, err)
	}
	second := recordTrunkCall(t, s, "task-asr", "call-after-edit")
	if err := s.ReserveExecute(second.DispatcherID, second.EventID, currentReservation(), at); !errors.Is(err, ErrNotReady) {
		t.Fatalf("old selection revision was accepted after edit: %v", err)
	}
	selection := currentReservation()
	selection.TaskRevision = 2
	if err := s.ReserveExecute(second.DispatcherID, second.EventID, selection, at); !errors.Is(err, ErrCapacity) {
		t.Fatalf("new lower task-line limit was ignored: %v", err)
	}
	if err := s.FinishExecute(first.DispatcherID, first.EventID); err != nil {
		t.Fatal(err)
	}
	if err := s.ReserveExecute(second.DispatcherID, second.EventID, selection, at); err != nil {
		t.Fatalf("capacity did not reopen after confirmed end: %v", err)
	}
}