package vaultik_test

import (
	"bytes"
	"context"
	"encoding/json"
	"strings"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"sneak.berlin/go/vaultik/internal/log"
	"sneak.berlin/go/vaultik/internal/snapshot"
)

// testBlobHashB is a blob that the manifest written by addRemote does
// not reference.
const testBlobHashB = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" +
	"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"

// TestRemoteInfo_UnreadableManifestLeavesOrphansUnknown checks that a
// manifest remote info cannot read makes the orphan figures unknown. A
// blob referenced only by that snapshot would otherwise be counted as
// orphaned, and the report would advise running prune.
func TestRemoteInfo_UnreadableManifestLeavesOrphansUnknown(t *testing.T) {
	log.Initialize(log.Config{})
	t.Parallel()

	env := newListEnv(t)

	// The readable manifest references blob A only.
	env.addRemote(t, listRemoteID, time.Date(2026, 3, 2, 0, 0, 0, 0, time.UTC))
	addBlob(t, env.store.testStorer, testBlobHashA)
	addBlob(t, env.store.testStorer, testBlobHashB)

	// With every manifest readable, blob B is orphaned.
	require.NoError(t, env.v.RemoteInfo(true))

	var doc map[string]any

	require.NoError(t, json.Unmarshal(env.stdout.Bytes(), &doc))
	assert.InDelta(t, 1, doc["orphaned_blob_count"], 0)

	// A second snapshot whose manifest cannot be decoded. Blob B may be
	// one of its blobs.
	unreadableKey := snapshot.RemoteSnapshotKey(listLocalID)
	require.NoError(t, env.store.Put(context.Background(),
		"metadata/"+unreadableKey+"/manifest.json.zst",
		bytes.NewReader([]byte("not a valid manifest"))))

	env.stdout.Reset()
	require.NoError(t, env.v.RemoteInfo(false))

	text := env.stdout.String()
	assert.Contains(t, text, "Orphaned (unreferenced):    unknown "+
		"(1 manifest(s) could not be read, "+
		"0 manifest(s) under a non-conforming name skipped)")
	assert.NotContains(t, text, "vaultik prune")

	env.stdout.Reset()
	require.NoError(t, env.v.RemoteInfo(true))

	doc = nil
	require.NoError(t, json.Unmarshal(env.stdout.Bytes(), &doc))
	assert.Contains(t, doc, "orphaned_blob_count")
	assert.Nil(t, doc["orphaned_blob_count"])
	assert.Contains(t, doc, "orphaned_blob_size")
	assert.Nil(t, doc["orphaned_blob_size"])
	assert.Equal(t, []any{unreadableKey}, doc["unreadable_manifests"])
}

// TestRemoteInfo_UnreadableManifestLeavesSnapshotBlobsUnknown checks
// that the row of a snapshot whose manifest cannot be read gives its
// blob count and blob size as unknown in the table and as null in
// --json, not as 0. A directory without a manifest still shows 0: the
// orphan figures count its blobs as orphaned, so it references none.
func TestRemoteInfo_UnreadableManifestLeavesSnapshotBlobsUnknown(t *testing.T) {
	log.Initialize(log.Config{})
	t.Parallel()

	env := newListEnv(t)
	readableKey := env.addRemote(t, listRemoteID,
		time.Date(2026, 3, 2, 0, 0, 0, 0, time.UTC))

	unreadableKey := snapshot.RemoteSnapshotKey(listLocalID)
	require.NoError(t, env.store.Put(context.Background(),
		"metadata/"+unreadableKey+"/manifest.json.zst",
		bytes.NewReader([]byte("not a valid manifest"))))

	noManifestKey := snapshot.RemoteSnapshotKey("testhost_home_2026-03-03T10:00:00Z")
	require.NoError(t, env.store.Put(context.Background(),
		"metadata/"+noManifestKey+"/db.zst.age",
		bytes.NewReader([]byte("not a valid database"))))

	require.NoError(t, env.v.RemoteInfo(false))

	// The table truncates the remote key, so a row is found by a prefix.
	wantUnknown := map[string]int{readableKey: 0, unreadableKey: 2, noManifestKey: 0}
	for key, want := range wantUnknown {
		var row string

		for line := range strings.SplitSeq(env.stdout.String(), "\n") {
			if strings.HasPrefix(line, key[:16]) {
				row = line
			}
		}

		require.NotEmpty(t, row, "no table row for %s", key)
		assert.Equal(t, want, strings.Count(row, "unknown"), "row: %q", row)
	}

	env.stdout.Reset()
	require.NoError(t, env.v.RemoteInfo(true))

	var doc struct {
		Snapshots []map[string]any `json:"snapshots"`
	}

	require.NoError(t, json.Unmarshal(env.stdout.Bytes(), &doc))
	require.Len(t, doc.Snapshots, len(wantUnknown))

	for _, entry := range doc.Snapshots {
		switch entry["snapshot_id"] {
		case readableKey:
			assert.InDelta(t, 1, entry["blob_count"], 0)
			assert.InDelta(t, fiveMegabytes, entry["blobs_size"], 0)
		case noManifestKey:
			assert.InDelta(t, 0, entry["blob_count"], 0)
			assert.InDelta(t, 0, entry["blobs_size"], 0)
		default:
			assert.Equal(t, unreadableKey, entry["snapshot_id"])
			assert.Contains(t, entry, "blob_count")
			assert.Nil(t, entry["blob_count"])
			assert.Contains(t, entry, "blobs_size")
			assert.Nil(t, entry["blobs_size"])
		}
	}
}

// TestRemoteInfo_SkipsNonConformingMetadataName checks that a directory
// under metadata/ whose name is not a remote key is left out of the
// report, and that the orphan figures are unknown when it holds a
// manifest. The name comes from the destination store; printed raw, its
// control characters would reach the terminal. Its manifest is not
// read, so a blob only it references would otherwise be counted as
// orphaned.
func TestRemoteInfo_SkipsNonConformingMetadataName(t *testing.T) {
	log.Initialize(log.Config{})
	t.Parallel()

	env := newListEnv(t)
	env.addRemote(t, listRemoteID, time.Date(2026, 3, 2, 0, 0, 0, 0, time.UTC))
	addBlob(t, env.store.testStorer, testBlobHashA)
	addBlob(t, env.store.testStorer, testBlobHashB)
	require.NoError(t, env.store.Put(context.Background(),
		"metadata/\x1b[31mred/manifest.json.zst",
		bytes.NewReader([]byte("not a valid manifest"))))

	require.NoError(t, env.v.RemoteInfo(false))

	text := env.stdout.String()
	assert.NotContains(t, text, "\x1b")
	assert.NotContains(t, text, "31mred")
	assert.Contains(t, text, "Total (1 snapshots)")
	assert.Contains(t, text, "Orphaned (unreferenced):    unknown "+
		"(0 manifest(s) could not be read, "+
		"1 manifest(s) under a non-conforming name skipped)")
	assert.NotContains(t, text, "vaultik prune")

	env.stdout.Reset()
	require.NoError(t, env.v.RemoteInfo(true))

	out := env.stdout.String()
	assert.NotContains(t, out, "31mred")

	var doc map[string]any

	require.NoError(t, json.Unmarshal([]byte(out), &doc))
	assert.Contains(t, doc, "orphaned_blob_count")
	assert.Nil(t, doc["orphaned_blob_count"])
	assert.Contains(t, doc, "orphaned_blob_size")
	assert.Nil(t, doc["orphaned_blob_size"])
	assert.InDelta(t, 1, doc["skipped_manifest_count"], 0)
	assert.NotContains(t, doc, "unreadable_manifests")
}

// TestRemoteInfo_DoesNotCountNonBlobFileUnderBlobs checks that a file
// under blobs/ whose name is not a blob hash is not counted as a blob.
// prune never deletes such a file, so counting it as an orphaned blob
// would advise a prune that removes nothing.
func TestRemoteInfo_DoesNotCountNonBlobFileUnderBlobs(t *testing.T) {
	log.Initialize(log.Config{})
	t.Parallel()

	env := newListEnv(t)
	env.addRemote(t, listRemoteID, time.Date(2026, 3, 2, 0, 0, 0, 0, time.UTC))
	addBlob(t, env.store.testStorer, testBlobHashA)
	require.NoError(t, env.store.Put(context.Background(),
		"blobs/"+testBlobHashA[:2]+"/"+testBlobHashA[2:4]+"/.DS_Store",
		bytes.NewReader([]byte("junk"))))

	require.NoError(t, env.v.RemoteInfo(false))

	text := env.stdout.String()
	assert.Contains(t, text, "Total blobs on remote:      1 (")
	assert.Contains(t, text, "Orphaned (unreferenced):    0 (")
	assert.NotContains(t, text, "vaultik prune")

	env.stdout.Reset()
	require.NoError(t, env.v.RemoteInfo(true))

	var doc map[string]any

	require.NoError(t, json.Unmarshal(env.stdout.Bytes(), &doc))
	assert.InDelta(t, 1, doc["total_blob_count"], 0)
	assert.InDelta(t, 0, doc["orphaned_blob_count"], 0)
	assert.InDelta(t, 0, doc["orphaned_blob_size"], 0)
}

// TestRemoteInfo_DirectoryWithoutManifestLeavesOrphansKnown checks that
// a directory under metadata/ holding no manifest.json.zst, such as one
// left by a backup interrupted before its manifest upload, leaves the
// orphan figures known. prune does not treat such a directory as a
// snapshot and deletes the blobs the report lists as orphaned.
func TestRemoteInfo_DirectoryWithoutManifestLeavesOrphansKnown(t *testing.T) {
	log.Initialize(log.Config{})
	t.Parallel()

	env := newListEnv(t)
	env.addRemote(t, listRemoteID, time.Date(2026, 3, 2, 0, 0, 0, 0, time.UTC))
	addBlob(t, env.store.testStorer, testBlobHashA)
	addBlob(t, env.store.testStorer, testBlobHashB)

	// One directory under a remote key and one under a non-conforming
	// name, each holding only a database.
	names := []string{snapshot.RemoteSnapshotKey(listLocalID), "\x1b[31mred"}
	for _, name := range names {
		require.NoError(t, env.store.Put(context.Background(),
			"metadata/"+name+"/db.zst.age",
			bytes.NewReader([]byte("not a valid database"))))
	}

	require.NoError(t, env.v.RemoteInfo(false))

	text := env.stdout.String()
	assert.NotContains(t, text, "\x1b")
	assert.Contains(t, text, "Downloading 1 manifest(s)...")
	assert.Contains(t, text, "Orphaned (unreferenced):    1 (")
	assert.Contains(t, text, "Run 'vaultik prune' to remove orphaned blobs.")

	env.stdout.Reset()
	require.NoError(t, env.v.RemoteInfo(true))

	var doc map[string]any

	require.NoError(t, json.Unmarshal(env.stdout.Bytes(), &doc))
	assert.InDelta(t, 1, doc["orphaned_blob_count"], 0)
	assert.NotContains(t, doc, "unreadable_manifests")
	assert.NotContains(t, doc, "skipped_manifest_count")
}