// Licensed to Apache Software Foundation (ASF) under one or more contributor
// license agreements. See the NOTICE file distributed with
// this work for additional information regarding copyright
// ownership. Apache Software Foundation (ASF) licenses this file to you under
// the Apache License, Version 2.0 (the "License"); you may
// not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

package backup

import (
	"context"
	"io"
	"os"
	"path/filepath"
	"strings"
	"testing"

	commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
	"github.com/apache/skywalking-banyandb/banyand/backup/snapshot"
	"github.com/apache/skywalking-banyandb/banyand/internal/storage"
	"github.com/apache/skywalking-banyandb/pkg/fs/remote/local"
)

const testTimeDir = "2023-10-10"

func TestRestoreDownload(t *testing.T) {
	remoteDir := t.TempDir()
	localRestoreDir := t.TempDir()

	fs, err := local.NewFS(remoteDir)
	if err != nil {
		t.Fatalf("failed to create remote FS: %v", err)
	}

	timeDir := testTimeDir
	remoteFilePath := filepath.Join(timeDir, snapshot.CatalogName(commonv1.Catalog_CATALOG_STREAM), "test.txt")
	content := "hello"
	err = fs.Upload(context.Background(), remoteFilePath, strings.NewReader(content))
	if err != nil {
		t.Fatalf("failed to upload file: %v", err)
	}

	err = restoreCatalog(fs, timeDir, localRestoreDir, commonv1.Catalog_CATALOG_STREAM)
	if err != nil {
		t.Fatalf("restoreCatalog failed: %v", err)
	}

	localFilePath := filepath.Join(localRestoreDir, snapshot.CatalogName(commonv1.Catalog_CATALOG_STREAM), storage.DataDir, "test.txt")
	got, err := os.ReadFile(localFilePath)
	if err != nil {
		t.Fatalf("failed to read local file: %v", err)
	}
	if string(got) != content {
		t.Fatalf("expected content %q, got %q", content, string(got))
	}
}

func TestRestoreDelete(t *testing.T) {
	remoteDir := t.TempDir()
	localRestoreDir := t.TempDir()

	fs, err := local.NewFS(remoteDir)
	if err != nil {
		t.Fatalf("failed to create remote FS: %v", err)
	}

	streamDir := filepath.Join(localRestoreDir, "stream", storage.DataDir)
	if err = os.MkdirAll(streamDir, storage.DirPerm); err != nil {
		t.Fatalf("failed to create local stream directory: %v", err)
	}
	extraFilePath := filepath.Join(streamDir, "old.txt")
	extraContent := "stale"
	if err = os.WriteFile(extraFilePath, []byte(extraContent), 0o600); err != nil {
		t.Fatalf("failed to write extra local file: %v", err)
	}

	timeDir := testTimeDir
	err = restoreCatalog(fs, timeDir, localRestoreDir, commonv1.Catalog_CATALOG_STREAM)
	if err != nil {
		t.Fatalf("restoreCatalog failed: %v", err)
	}

	if _, err := os.Stat(extraFilePath); !os.IsNotExist(err) {
		t.Fatalf("expected extra file %q to be deleted", extraFilePath)
	}
}

func TestRestoreSchemaProperty(t *testing.T) {
	remoteDir := t.TempDir()
	localRestoreDir := t.TempDir()

	fs, err := local.NewFS(remoteDir)
	if err != nil {
		t.Fatalf("failed to create remote FS: %v", err)
	}

	timeDir := testTimeDir
	remoteFilePath := filepath.Join(timeDir, snapshot.SchemaPropertyCatalogName, "test.txt")
	content := "schema-hello"
	err = fs.Upload(context.Background(), remoteFilePath, strings.NewReader(content))
	if err != nil {
		t.Fatalf("failed to upload file: %v", err)
	}

	err = restoreByName(fs, timeDir, localRestoreDir, snapshot.SchemaPropertyCatalogName)
	if err != nil {
		t.Fatalf("restoreByName failed: %v", err)
	}

	localFilePath := filepath.Join(localRestoreDir, snapshot.SchemaPropertyCatalogName, storage.DataDir, "test.txt")
	got, err := os.ReadFile(localFilePath)
	if err != nil {
		t.Fatalf("failed to read local file: %v", err)
	}
	if string(got) != content {
		t.Fatalf("expected content %q, got %q", content, string(got))
	}
}

func TestRestoreSchemaPropertyDelete(t *testing.T) {
	remoteDir := t.TempDir()
	localRestoreDir := t.TempDir()

	fs, err := local.NewFS(remoteDir)
	if err != nil {
		t.Fatalf("failed to create remote FS: %v", err)
	}

	schemaDir := filepath.Join(localRestoreDir, snapshot.SchemaPropertyCatalogName, storage.DataDir)
	if err = os.MkdirAll(schemaDir, storage.DirPerm); err != nil {
		t.Fatalf("failed to create local schema-property directory: %v", err)
	}
	extraFilePath := filepath.Join(schemaDir, "old.txt")
	if err = os.WriteFile(extraFilePath, []byte("stale"), 0o600); err != nil {
		t.Fatalf("failed to write extra local file: %v", err)
	}

	timeDir := testTimeDir
	err = restoreByName(fs, timeDir, localRestoreDir, snapshot.SchemaPropertyCatalogName)
	if err != nil {
		t.Fatalf("restoreByName failed: %v", err)
	}

	if _, err := os.Stat(extraFilePath); !os.IsNotExist(err) {
		t.Fatalf("expected extra file %q to be deleted", extraFilePath)
	}
}

func TestRestoreSame(t *testing.T) {
	remoteDir := t.TempDir()
	localRestoreDir := t.TempDir()

	fs, err := local.NewFS(remoteDir)
	if err != nil {
		t.Fatalf("failed to create remote FS: %v", err)
	}

	streamDir := filepath.Join(localRestoreDir, "stream", storage.DataDir)
	if err = os.MkdirAll(streamDir, storage.DirPerm); err != nil {
		t.Fatalf("failed to create local stream directory: %v", err)
	}
	extraFilePath := filepath.Join(streamDir, "test.txt")
	extraContent := "hello"
	if err = os.WriteFile(extraFilePath, []byte(extraContent), 0o600); err != nil {
		t.Fatalf("failed to write extra local file: %v", err)
	}

	timeDir := testTimeDir
	remoteFilePath := filepath.Join(timeDir, snapshot.CatalogName(commonv1.Catalog_CATALOG_STREAM), "test.txt")
	content := "hello"
	err = fs.Upload(context.Background(), remoteFilePath, strings.NewReader(content))
	if err != nil {
		t.Fatalf("failed to upload file: %v", err)
	}

	err = restoreCatalog(fs, timeDir, localRestoreDir, commonv1.Catalog_CATALOG_STREAM)
	if err != nil {
		t.Fatalf("restoreCatalog failed: %v", err)
	}

	_, err = os.Stat(extraFilePath)
	if err != nil {
		t.Fatalf("expected extra file %q exist", extraFilePath)
	}
}

func TestRestoreRejectsRemotePathTraversal(t *testing.T) {
	timeDir := testTimeDir
	catalogName := snapshot.CatalogName(commonv1.Catalog_CATALOG_STREAM)
	localRestoreDir := t.TempDir()
	escapedFile := filepath.Join(localRestoreDir, catalogName, "escaped.txt")
	fs := &restoreTraversalFS{
		files: []string{
			filepath.ToSlash(filepath.Join(timeDir, catalogName, "..", "escaped.txt")),
		},
	}

	err := restoreByName(fs, timeDir, localRestoreDir, catalogName)
	if err == nil {
		t.Fatal("expected restoreByName to reject remote path traversal")
	}
	if _, statErr := os.Stat(escapedFile); !os.IsNotExist(statErr) {
		t.Fatalf("escaped file exists or stat failed with unexpected error: %v", statErr)
	}
}

func TestValidatedRemoteRelPath(t *testing.T) {
	timeDir := testTimeDir
	catalogName := snapshot.CatalogName(commonv1.Catalog_CATALOG_STREAM)
	validRemoteFile := filepath.ToSlash(filepath.Join(timeDir, catalogName, "nested", "test.txt"))

	relPath, err := validatedRemoteRelPath(timeDir, catalogName, validRemoteFile)
	if err != nil {
		t.Fatalf("validatedRemoteRelPath failed: %v", err)
	}
	if relPath != "nested/test.txt" {
		t.Fatalf("relPath = %q, want %q", relPath, "nested/test.txt")
	}
}

func TestValidatedRemoteRelPathRejectsInvalidPaths(t *testing.T) {
	timeDir := testTimeDir
	catalogName := snapshot.CatalogName(commonv1.Catalog_CATALOG_STREAM)
	tests := []struct {
		remoteFile string
		name       string
	}{
		{
			name:       "absolute path",
			remoteFile: "/tmp/backup/file.txt",
		},
		{
			name:       "catalog prefix only",
			remoteFile: filepath.ToSlash(filepath.Join(timeDir, catalogName)),
		},
		{
			name:       "outside catalog prefix",
			remoteFile: filepath.ToSlash(filepath.Join(timeDir, "measure", "test.txt")),
		},
		{
			name:       "parent traversal",
			remoteFile: filepath.ToSlash(filepath.Join(timeDir, catalogName, "..", "escaped.txt")),
		},
		{
			name:       "volume name",
			remoteFile: `C:/backup/file.txt`,
		},
	}

	for _, testCase := range tests {
		t.Run(testCase.name, func(t *testing.T) {
			if _, err := validatedRemoteRelPath(timeDir, catalogName, testCase.remoteFile); err == nil {
				t.Fatal("expected invalid remote path to be rejected")
			}
		})
	}
}

type restoreTraversalFS struct {
	files []string
}

func (r *restoreTraversalFS) Upload(_ context.Context, _ string, _ io.Reader) error {
	return nil
}

func (r *restoreTraversalFS) Download(_ context.Context, _ string) (io.ReadCloser, error) {
	return io.NopCloser(strings.NewReader("escape")), nil
}

func (r *restoreTraversalFS) List(_ context.Context, _ string) ([]string, error) {
	return r.files, nil
}

func (r *restoreTraversalFS) Delete(_ context.Context, _ string) error {
	return nil
}

func (r *restoreTraversalFS) Close() error {
	return nil
}
