kenn-io--agentsview/cmd/agentsview/archive_write_backend_test.go

package main

import (
	"bytes"
	"context"
	"path/filepath"
	"strings"
	stdsync "sync"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"go.kenn.io/agentsview/internal/config"
	"go.kenn.io/agentsview/internal/db"
	"go.kenn.io/agentsview/internal/dbtest"
	"go.kenn.io/agentsview/internal/postgres"
	syncpkg "go.kenn.io/agentsview/internal/sync"
)

// TestDaemonPushHeartbeat pins the daemon-delegated push's client-side
// output: an immediate delegation announcement, elapsed-time heartbeats
// while the blocking POST is in flight, and silence after stop.
func TestDaemonPushHeartbeat(t *testing.T) {
	orig := daemonPushHeartbeatInterval
	daemonPushHeartbeatInterval = 5 * time.Millisecond
	t.Cleanup(func() { daemonPushHeartbeatInterval = orig })

	var out syncBuffer
	stop := startDaemonPushHeartbeatTo(&out, "PostgreSQL")
	assert.Contains(t, out.String(),
		"Pushing to PostgreSQL via the local daemon")

	require.Eventually(t, func() bool {
		return strings.Contains(out.String(),
			"still pushing to PostgreSQL via the daemon")
	}, time.Second, time.Millisecond, "heartbeat line must appear")

	stop()
	settled := out.String()
	time.Sleep(20 * time.Millisecond)
	assert.Equal(t, settled, out.String(), "no output after stop")
}

// TestDaemonPushProgressSilencesHeartbeat pins that the first streamed
// progress event stops the elapsed-time heartbeat for good: once the daemon
// starts reporting real progress, heartbeat lines must not interleave with
// the in-place progress line.
func TestDaemonPushProgressSilencesHeartbeat(t *testing.T) {
	orig := daemonPushHeartbeatInterval
	daemonPushHeartbeatInterval = 50 * time.Millisecond
	t.Cleanup(func() { daemonPushHeartbeatInterval = orig })

	out := captureStdout(t, func() {
		onProgress, finish := daemonPushProgress(
			"PostgreSQL", newPGPushProgressPrinter(),
		)
		onProgress(postgres.PushProgress{SessionsDone: 1, SessionsTotal: 2})
		time.Sleep(150 * time.Millisecond)
		finish()
	})

	assert.Contains(t, out, "Pushing to PostgreSQL via the local daemon")
	assert.Contains(t, out, "Pushing... 1/2 sessions")
	assert.NotContains(t, out, "still pushing",
		"heartbeat must stay silent after the first progress event")
}

// syncBuffer is a mutex-guarded bytes.Buffer: the heartbeat goroutine writes
// while the test reads.
type syncBuffer struct {
	mu  stdsync.Mutex
	buf bytes.Buffer
}

func (b *syncBuffer) Write(p []byte) (int, error) {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.buf.Write(p)
}

func (b *syncBuffer) String() string {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.buf.String()
}

func TestLocalArchiveWriteBackendPGPushStopsAfterCanceledLocalSync(t *testing.T) {
	testLocalArchivePushStopsAfterCanceledSync(t,
		func(backend *localArchiveWriteBackend, ctx context.Context) error {
			_, err := backend.PGPush(
				ctx, pgTargetSelection{}, PGPushConfig{}, nil, nil,
			)
			return err
		})
}

func TestLocalArchiveWriteBackendDuckDBPushStopsAfterCanceledLocalSync(t *testing.T) {
	testLocalArchivePushStopsAfterCanceledSync(t,
		func(backend *localArchiveWriteBackend, ctx context.Context) error {
			_, err := backend.DuckDBPush(
				ctx, config.DuckDBConfig{}, DuckDBPushConfig{}, nil, nil,
			)
			return err
		})
}

func TestLocalArchiveWriteBackendDuckDBPushUsesConfiguredRemoteURL(t *testing.T) {
	backend := testLocalArchiveWriteBackend(t)

	captureStdout(t, func() {
		_, err := backend.DuckDBPush(
			context.Background(),
			config.DuckDBConfig{
				URL:         "quack:https://duck.example.test",
				MachineName: "workstation",
			},
			DuckDBPushConfig{},
			nil,
			nil,
		)
		require.Error(t, err)
		assert.Contains(t, err.Error(), "duckdb quack token is required")
	})
}

func TestLocalArchiveWriteBackendDuckDBPushValidatesRemoteBeforeLocalSync(t *testing.T) {
	backend := testLocalArchiveWriteBackend(t)

	var err error
	out := captureStdout(t, func() {
		_, err = backend.DuckDBPush(
			context.Background(),
			config.DuckDBConfig{
				URL:         "quack:https://duck.example.test",
				MachineName: "workstation",
			},
			DuckDBPushConfig{Full: true},
			nil,
			nil,
		)
	})

	require.Error(t, err)
	assert.Contains(t, err.Error(), "duckdb quack token is required")
	assert.NotContains(t, out, "Database:")
	assert.NotContains(t, out, "Opening DuckDB mirror")
}

func TestRunPGWatchStartupSyncFallsBackAfterAbortedResync(t *testing.T) {
	database := dbtest.OpenTestDB(t)
	missingPath := filepath.Join(t.TempDir(), "missing.jsonl")
	dbtest.SeedSession(t, database, "existing", "proj",
		func(s *db.Session) {
			s.FilePath = &missingPath
		})
	engine := syncpkg.NewEngine(database, syncpkg.EngineConfig{})

	didResync, err := runPGWatchStartupSync(
		context.Background(), engine, true,
	)

	require.NoError(t, err)
	assert.True(t, didResync)
	assert.False(t, engine.LastSyncStats().Aborted)
	assert.False(t, engine.LastSync().IsZero())
}

func TestLocalArchiveWriteBackendPGPushWatchCanceledStartupIsClean(t *testing.T) {
	backend := testLocalArchiveWriteBackend(t)

	err := backend.PGPushWatch(
		canceledContext(),
		pgTargetSelection{},
		PGPushConfig{},
		nil,
		nil,
		time.Millisecond,
		time.Millisecond,
	)

	require.NoError(t, err)
}

func TestResolveArchiveWriteBackendCopiesNoSyncRuntime(t *testing.T) {
	dataDir := t.TempDir()
	host, port := testPingServer(t)
	_, err := WriteDaemonRuntimeWithAuthAndNoSync(
		dataDir, host, port, "test", false, false, true,
	)
	require.NoError(t, err)
	t.Cleanup(func() { RemoveDaemonRuntime(dataDir) })

	backend, cleanup, err := resolveArchiveWriteBackend(
		context.Background(), config.Config{DataDir: dataDir},
	)
	require.NoError(t, err)
	t.Cleanup(cleanup)

	daemonBackend, ok := backend.(daemonArchiveWriteBackend)
	require.True(t, ok)
	assert.True(t, daemonBackend.appCfg.NoSync)
}

// canceledContext returns a context that has already been canceled.
func canceledContext() context.Context {
	ctx, cancel := context.WithCancel(context.Background())
	cancel()
	return ctx
}

// testLocalArchivePushStopsAfterCanceledSync asserts that a local push aborts
// with context.Canceled when its context is already canceled. push runs the
// backend-specific push call and returns its error.
func testLocalArchivePushStopsAfterCanceledSync(
	t *testing.T,
	push func(*localArchiveWriteBackend, context.Context) error,
) {
	t.Helper()
	backend := testLocalArchiveWriteBackend(t)

	var err error
	captureStdout(t, func() {
		err = push(backend, canceledContext())
	})

	require.ErrorIs(t, err, context.Canceled)
}

func testLocalArchiveWriteBackend(t *testing.T) *localArchiveWriteBackend {
	t.Helper()
	dataDir := t.TempDir()
	dbPath := filepath.Join(dataDir, "sessions.db")
	database := dbtest.OpenTestDBAt(t, dbPath)

	return &localArchiveWriteBackend{
		appCfg: config.Config{
			DataDir: dataDir,
			DBPath:  dbPath,
		},
		database: database,
	}
}