teraWeb/backend/internal/directstream/ticket_socket_test.go

package directstream

import (
	"context"
	"errors"
	"net"
	"os"
	"path/filepath"
	"reflect"
	"sync"
	"testing"
	"time"
)

func TestListenTicketSocketRequiresAbsolutePathAndRefusesNonSocket(t *testing.T) {
	if listener, err := ListenTicketSocket("relative.sock"); err == nil {
		listener.Close()
		t.Fatal("expected relative ticket socket path rejection")
	}
	path := filepath.Join(t.TempDir(), "direct-tickets.sock")
	if err := os.WriteFile(path, []byte("do not replace"), 0o600); err != nil {
		t.Fatal(err)
	}
	if listener, err := ListenTicketSocket(path); err == nil {
		listener.Close()
		t.Fatal("expected non-socket path rejection")
	}
}

type testTicketAuthority struct {
	store    *TicketStore
	mu       sync.Mutex
	closed   []TicketClaims
	released []TicketClaims
	events   []SessionEvent
}

func (a *testTicketAuthority) ConsumeTicket(_ context.Context, value string, binding TicketBinding) (TicketClaims, error) {
	return a.store.Consume(value, binding)
}

func (a *testTicketAuthority) CloseConsumedStream(_ context.Context, claims TicketClaims) error {
	a.mu.Lock()
	defer a.mu.Unlock()
	a.closed = append(a.closed, claims)
	return nil
}

func (a *testTicketAuthority) ReleaseConsumedWriter(_ context.Context, claims TicketClaims) error {
	a.mu.Lock()
	defer a.mu.Unlock()
	a.released = append(a.released, claims)
	return nil
}

func (a *testTicketAuthority) RecordConsumedStreamEvent(_ context.Context, _ TicketClaims, event SessionEvent) error {
	a.mu.Lock()
	defer a.mu.Unlock()
	a.events = append(a.events, event)
	return nil
}

func TestTicketSocketConsumesOnceAndKeepsClaimsServerSideForLifecycle(t *testing.T) {
	authority := &testTicketAuthority{store: NewTicketStore(TicketStoreOptions{})}
	socketPath, stop := startTicketSocketServer(t, authority)
	defer stop()
	client := NewTicketSocketClient(socketPath)
	claims := TicketClaims{SessionID: "direct-CH01-1", Channel: "CH01", Actor: "admin", Access: AccessWriter}
	issued, err := authority.store.Issue(claims)
	if err != nil {
		t.Fatal(err)
	}

	got, err := client.ConsumeTicket(context.Background(), issued.Value, TicketBinding{
		SessionID: claims.SessionID, Channel: claims.Channel,
	})
	if err != nil {
		t.Fatalf("consume: %v", err)
	}
	if !reflect.DeepEqual(got, claims) {
		t.Fatalf("claims = %#v, want %#v", got, claims)
	}
	event := SessionEvent{Type: SessionEventConnected, Source: "192.0.2.10", Timestamp: time.Now().UTC()}
	if err := client.RecordDirectStreamEvent(context.Background(), TicketBinding{
		SessionID: claims.SessionID, Channel: claims.Channel,
	}, event); err != nil {
		t.Fatalf("record session event: %v", err)
	}
	if _, err := client.ConsumeTicket(context.Background(), issued.Value, TicketBinding{
		SessionID: claims.SessionID, Channel: claims.Channel,
	}); !errors.Is(err, ErrInvalidTicket) {
		t.Fatalf("replay error = %v", err)
	}
	if err := client.ReleaseDirectWriter(context.Background(), TicketBinding{
		SessionID: claims.SessionID, Channel: claims.Channel,
	}); err != nil {
		t.Fatalf("release writer: %v", err)
	}
	if err := client.CloseDirectStream(context.Background(), TicketBinding{
		SessionID: claims.SessionID, Channel: claims.Channel,
	}); err != nil {
		t.Fatalf("close: %v", err)
	}
	if !reflect.DeepEqual(authority.released, []TicketClaims{claims}) || !reflect.DeepEqual(authority.closed, []TicketClaims{claims}) || !reflect.DeepEqual(authority.events, []SessionEvent{event}) {
		t.Fatalf("released = %#v, closed = %#v, events = %#v", authority.released, authority.closed, authority.events)
	}
	if err := client.CloseDirectStream(context.Background(), TicketBinding{
		SessionID: claims.SessionID, Channel: claims.Channel,
	}); err == nil {
		t.Fatal("expected repeated close to fail closed")
	}
}

func TestTicketSocketRejectsWrongBindingWithoutConsumingTicket(t *testing.T) {
	authority := &testTicketAuthority{store: NewTicketStore(TicketStoreOptions{})}
	socketPath, stop := startTicketSocketServer(t, authority)
	defer stop()
	client := NewTicketSocketClient(socketPath)
	claims := TicketClaims{SessionID: "direct-CH01-2", Channel: "CH01", Actor: "operator", Access: AccessViewer}
	issued, err := authority.store.Issue(claims)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := client.ConsumeTicket(context.Background(), issued.Value, TicketBinding{
		SessionID: claims.SessionID, Channel: "CH02",
	}); !errors.Is(err, ErrInvalidTicket) {
		t.Fatalf("wrong binding error = %v", err)
	}
	if _, err := client.ConsumeTicket(context.Background(), issued.Value, TicketBinding{
		SessionID: claims.SessionID, Channel: claims.Channel,
	}); err != nil {
		t.Fatalf("correct binding after rejection: %v", err)
	}
	if err := client.CloseDirectStream(context.Background(), TicketBinding{
		SessionID: "forged", Channel: claims.Channel,
	}); err == nil {
		t.Fatal("expected lifecycle operation for unconsumed binding to fail")
	}
}

func TestTicketSocketHeartbeatStatusAndInactiveCleanup(t *testing.T) {
	now := time.Date(2026, 7, 22, 12, 0, 0, 0, time.UTC)
	authority := &testTicketAuthority{store: NewTicketStore(TicketStoreOptions{Clock: func() time.Time { return now }})}
	server := &TicketSocketServer{
		Authority: authority,
		Clock:     func() time.Time { return now },
		ActiveTTL: 30 * time.Second,
	}
	socketPath := filepath.Join(t.TempDir(), "direct-tickets.sock")
	listener, err := net.Listen("unix", socketPath)
	if err != nil {
		t.Skipf("unix sockets unavailable: %v", err)
	}
	done := make(chan error, 1)
	go func() { done <- server.Serve(listener) }()
	defer func() {
		_ = listener.Close()
		<-done
	}()
	client := NewTicketSocketClient(socketPath)
	claims := TicketClaims{SessionID: "direct-CH01-heartbeat", Channel: "CH01", Actor: "admin", Access: AccessWriter}
	issued, err := authority.store.Issue(claims)
	if err != nil {
		t.Fatal(err)
	}
	binding := TicketBinding{SessionID: claims.SessionID, Channel: claims.Channel}
	if _, err := client.ConsumeTicket(context.Background(), issued.Value, binding); err != nil {
		t.Fatal(err)
	}
	if err := client.RecordDirectStreamEvent(context.Background(), binding, SessionEvent{
		Type: SessionEventConnected, Source: "not-an-ip", Timestamp: now,
	}); err == nil {
		t.Fatal("expected invalid session event source rejection")
	}
	if err := client.RecordDirectStreamEvent(context.Background(), binding, SessionEvent{
		Type: SessionEventConnected, Source: "198.51.100.20", Timestamp: now,
	}); err != nil {
		t.Fatalf("record session event: %v", err)
	}
	now = now.Add(20 * time.Second)
	if err := client.HeartbeatDirectStream(context.Background(), binding); err != nil {
		t.Fatalf("heartbeat: %v", err)
	}
	now = now.Add(20 * time.Second)
	if expired := server.ExpireInactive(context.Background()); expired != 0 {
		t.Fatalf("ExpireInactive() = %d after heartbeat, want 0", expired)
	}
	status, err := client.Status(context.Background())
	if err != nil {
		t.Fatalf("status: %v", err)
	}
	if status.ActiveBindings != 1 || status.Consumed != 1 || status.Heartbeats != 1 || status.SessionEvents != 1 {
		t.Fatalf("status = %#v", status)
	}

	now = now.Add(31 * time.Second)
	if expired := server.ExpireInactive(context.Background()); expired != 1 {
		t.Fatalf("ExpireInactive() = %d, want 1", expired)
	}
	status, err = client.Status(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	if status.ActiveBindings != 0 || status.ExpiredStreams != 1 {
		t.Fatalf("expired status = %#v", status)
	}
	if !reflect.DeepEqual(authority.closed, []TicketClaims{claims}) {
		t.Fatalf("expired claims closed = %#v", authority.closed)
	}
}

func startTicketSocketServer(t *testing.T, authority TicketAuthority) (string, func()) {
	t.Helper()
	socketPath := filepath.Join(t.TempDir(), "direct-tickets.sock")
	listener, err := net.Listen("unix", socketPath)
	if err != nil {
		t.Skipf("unix sockets unavailable: %v", err)
	}
	server := TicketSocketServer{Authority: authority}
	done := make(chan error, 1)
	go func() { done <- server.Serve(listener) }()
	return socketPath, func() {
		_ = listener.Close()
		select {
		case err := <-done:
			if err != nil {
				t.Errorf("ticket server stop: %v", err)
			}
		case <-time.After(time.Second):
			t.Error("ticket server did not stop")
		}
	}
}