package metadata

import (
	"context"
	"errors"
	"testing"

	apperrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/shared/pkg/models"
)

// reconcileMetaRepo is a MetadataRepository whose object metadata is always
// absent — modeling a delete retry after the object's metadata was already
// committed-removed (#3484).
type reconcileMetaRepo struct {
	ports.MetadataRepository
}

func (r *reconcileMetaRepo) GetSchemaByAPIName(_ context.Context, _ string) (*models.ObjectMetadata, error) {
	return nil, nil
}

// GetAllSchemas and GetAllRelationships back the ReadSource-corroboration read
// path (shouldReconcileOrphanTable → GetSchemas → core cache build). No live
// objects here, so the reconcile is never blocked by an aliased read table.
func (r *reconcileMetaRepo) GetAllSchemas(_ context.Context) ([]*models.ObjectMetadata, error) {
	return nil, nil
}

func (r *reconcileMetaRepo) GetAllRelationships(_ context.Context) ([]*models.Relationship, error) {
	return nil, nil
}

// reconcileSchemaRepo records DropTable calls so a test can assert whether the
// stranded-table reconcile fired. It reports the _System_Table ownership
// tombstone via HasDropPendingTombstone (default present) so the happy-path
// reconcile — a genuine phase-2 orphan, which always keeps its registry row —
// passes the provenance proof. Tests that model a non-platform table (raw
// operator/ETL table, or a same-shape CREATE TABLE ... LIKE copy) set
// notOwned=true: no registry row, no destruction.
type reconcileSchemaRepo struct {
	ports.SchemaRepository
	dropped  []string
	notOwned bool
	probeErr error
}

func (r *reconcileSchemaRepo) DropTable(tableName string) error {
	r.dropped = append(r.dropped, tableName)
	return nil
}

// HasDropPendingTombstone models the ownership-tombstone probe. Default is true
// (the reconcile target is a genuine phase-2 orphan carrying its registry row);
// notOwned models any non-platform table sharing the name; probeErr models a
// probe failure.
func (r *reconcileSchemaRepo) HasDropPendingTombstone(_ string) (bool, error) {
	if r.probeErr != nil {
		return false, r.probeErr
	}
	return !r.notOwned, nil
}

// TestDeleteSchema_ReconcilesOrphanTableWhenMetadataAbsent verifies that a
// DeleteSchema retry, after the object's metadata is already gone, still calls
// DropTable to clear any stranded physical table — so a re-create
// (CREATE TABLE IF NOT EXISTS) cannot silently rebind to the orphan.
func TestDeleteSchema_ReconcilesOrphanTableWhenMetadataAbsent(t *testing.T) {
	sr := &reconcileSchemaRepo{}
	ms := NewMetadataService(&reconcileMetaRepo{}, NewSchemaManager(sr), nil)

	err := ms.DeleteSchema(context.Background(), "Widget__c")
	if !apperrors.IsNotFound(err) {
		t.Fatalf("DeleteSchema error = %T %v, want NotFound", err, err)
	}
	if len(sr.dropped) != 1 || sr.dropped[0] != "Widget__c" {
		t.Fatalf("DropTable calls = %v, want [Widget__c] (stranded table must be reconciled)", sr.dropped)
	}
}

// TestDeleteSchema_NeverDropsSystemTableOnReconcile is the safety guard: the
// orphan reconcile must NEVER drop a _System_* infrastructure table, even when
// no object metadata resolves for that name.
func TestDeleteSchema_NeverDropsSystemTableOnReconcile(t *testing.T) {
	sr := &reconcileSchemaRepo{}
	ms := NewMetadataService(&reconcileMetaRepo{}, NewSchemaManager(sr), nil)

	err := ms.DeleteSchema(context.Background(), "_System_Config")
	if !apperrors.IsNotFound(err) {
		t.Fatalf("DeleteSchema error = %T %v, want NotFound", err, err)
	}
	if len(sr.dropped) != 0 {
		t.Fatalf("DropTable calls = %v, want none (a _System_* table must never be dropped)", sr.dropped)
	}
}

// TestDeleteSchema_ReconcileNeverDropsManagedSubstrate is test-plan (8): the
// reconcile guard must refuse UNPREFIXED managed substrate (User, Profile,
// ObjectPermissions, ...) — the exact class the old raw HasPrefix(_System_)
// check missed. Each of these has no _System_Object row, so without the
// IsManagedSystemTable guard a DELETE would route straight into DropTable and
// destroy a live substrate table (#3484 blocker/fourth-miss).
func TestDeleteSchema_ReconcileNeverDropsManagedSubstrate(t *testing.T) {
	// Names delivered exactly as the interactive delete route delivers them,
	// including the route-lowercased forms that TiDB (lower_case_table_names=2)
	// would still resolve to the real table.
	for _, name := range []string{
		"ObjectPermissions", "objectpermissions",
		"FieldPermissions", "User", "Profile",
		"_System_InstalledPack", "_system_installedpack",
		"_System_FlowAuditLog", "_system_flowauditlog",
		"_System_NamedCredentialSecret",
	} {
		t.Run(name, func(t *testing.T) {
			sr := &reconcileSchemaRepo{}
			ms := NewMetadataService(&reconcileMetaRepo{}, NewSchemaManager(sr), nil)

			err := ms.DeleteSchema(context.Background(), name)
			if !apperrors.IsNotFound(err) {
				t.Fatalf("DeleteSchema(%q) error = %T %v, want NotFound", name, err, err)
			}
			if len(sr.dropped) != 0 {
				t.Fatalf("DeleteSchema(%q) DropTable calls = %v, want none (managed substrate must never be dropped)", name, sr.dropped)
			}
		})
	}
}

// readSourceAliasMetaRepo has no metadata under the queried name, but exposes a
// LIVE object whose ReadSource physically reads from that name — modeling a
// ReadSource-aliased table. Dropping it would destroy a live object's data.
type readSourceAliasMetaRepo struct {
	ports.MetadataRepository
	aliasPhysicalTable string
}

func (r *readSourceAliasMetaRepo) GetSchemaByAPIName(_ context.Context, _ string) (*models.ObjectMetadata, error) {
	return nil, nil
}

func (r *readSourceAliasMetaRepo) GetAllSchemas(_ context.Context) ([]*models.ObjectMetadata, error) {
	return []*models.ObjectMetadata{
		{
			APIName:    "AliasedLogical__c",
			ReadSource: &models.ObjectReadSourceConfig{PhysicalTable: r.aliasPhysicalTable},
		},
	}, nil
}

func (r *readSourceAliasMetaRepo) GetAllRelationships(_ context.Context) ([]*models.Relationship, error) {
	return nil, nil
}

// TestDeleteSchema_ReconcileRefusesReadSourceAliasedTable is test-plan hole #3:
// a physical table can legitimately back a LIVE object under a different logical
// name via ReadSource aliasing. When a live object resolves its read table to
// the delete target, the "no metadata ⇒ droppable orphan" assumption is false —
// the reconcile must refuse to drop, protecting the live object's data.
func TestDeleteSchema_ReconcileRefusesReadSourceAliasedTable(t *testing.T) {
	sr := &reconcileSchemaRepo{}
	ms := NewMetadataService(&readSourceAliasMetaRepo{aliasPhysicalTable: "Legacy_Widget_Table"}, NewSchemaManager(sr), nil)

	err := ms.DeleteSchema(context.Background(), "Legacy_Widget_Table")
	if !apperrors.IsNotFound(err) {
		t.Fatalf("DeleteSchema error = %T %v, want NotFound", err, err)
	}
	if len(sr.dropped) != 0 {
		t.Fatalf("DropTable calls = %v, want none (a ReadSource-aliased live table must never be dropped)", sr.dropped)
	}
}

// TestDeleteSchema_ReconcileDropError_PassesConflictThrough verifies that when
// the reconcile DropTable returns a typed ConflictError (e.g. an inbound-FK
// block naming the constraint), DeleteSchema passes it through unwrapped rather
// than sanitizing it into an opaque InternalError — so the remediation reaches
// the client.
func TestDeleteSchema_ReconcileDropError_PassesConflictThrough(t *testing.T) {
	sr := &conflictDropSchemaRepo{}
	ms := NewMetadataService(&reconcileMetaRepo{}, NewSchemaManager(sr), nil)

	err := ms.DeleteSchema(context.Background(), "Widget__c")
	var conflict *apperrors.ConflictError
	if !errors.As(err, &conflict) {
		t.Fatalf("DeleteSchema error = %T %v, want *ConflictError passed through unwrapped", err, err)
	}
}

// conflictDropSchemaRepo returns a ConflictError from DropTable. It reports a
// _System_Table registry tombstone so the reconcile passes the provenance proof
// and actually reaches DropTable (whose ConflictError we assert).
type conflictDropSchemaRepo struct {
	ports.SchemaRepository
}

func (r *conflictDropSchemaRepo) DropTable(tableName string) error {
	return apperrors.NewConflictErrorWithMessage(tableName, "blocked by fk_child_ref on Child__c")
}

func (r *conflictDropSchemaRepo) HasDropPendingTombstone(_ string) (bool, error) {
	return true, nil
}

// cacheLoadFailMetaRepo has absent object metadata (routing DeleteSchema into
// the no-live-metadata reconcile branch) but FAILS the core-cache build:
// GetAllSchemas returns an error, exactly the transient snapshot-build failure
// the ReadSource-corroboration path must survive. GetSchemas would swallow that
// into an empty slice — so without the ensureCoreCacheInitialized fail-closed
// gate, the reconcile would see zero live objects and DROP the table.
type cacheLoadFailMetaRepo struct {
	ports.MetadataRepository
}

func (r *cacheLoadFailMetaRepo) GetSchemaByAPIName(_ context.Context, _ string) (*models.ObjectMetadata, error) {
	return nil, nil
}

func (r *cacheLoadFailMetaRepo) GetAllSchemas(_ context.Context) ([]*models.ObjectMetadata, error) {
	return nil, errors.New("transient snapshot build failure")
}

func (r *cacheLoadFailMetaRepo) GetAllRelationships(_ context.Context) ([]*models.Relationship, error) {
	return nil, nil
}

// TestDeleteSchema_ReconcileRefusesOnCacheLoadFailure is the load-bearing guard
// for review finding #1 (the fail-open reconcile): when the schema cache cannot
// load, GetSchemas returns an EMPTY slice, which — without the fail-closed
// ensureCoreCacheInitialized gate in shouldReconcileOrphanTable — would let the
// reconcile DROP a table that may be the physical ReadSource of a LIVE object.
// Cold cache is the norm (invalidateCacheLocked fires after every DropTable), so
// this path is routinely reachable. The reconcile MUST refuse: no DropTable,
// NotFound returned, table preserved.
//
// Load-bearing: remove the ensureCoreCacheInitialized guard from
// shouldReconcileOrphanTable and this fails — GetSchemas silently returns [] and
// the reconcile authorizes the drop.
func TestDeleteSchema_ReconcileRefusesOnCacheLoadFailure(t *testing.T) {
	sr := &reconcileSchemaRepo{}
	ms := NewMetadataService(&cacheLoadFailMetaRepo{}, NewSchemaManager(sr), nil)

	err := ms.DeleteSchema(context.Background(), "Widget__c")
	if !apperrors.IsNotFound(err) {
		t.Fatalf("DeleteSchema error = %T %v, want NotFound", err, err)
	}
	if len(sr.dropped) != 0 {
		t.Fatalf("DropTable calls = %v, want none (a cache-load failure must fail closed, never drop)", sr.dropped)
	}
}

// TestDeleteSchema_ReconcileRefusesRawNonPlatformTable is the load-bearing guard
// for review finding #1 (D10): the reconcile authorizes a DROP for any
// metadata-absent name, but "no metadata" does NOT prove the physical table was
// ever a platform object. A raw operator/ETL/backup table sharing the name
// (noShape=true, i.e. no Id/CreatedDate/IsDeleted) must be preserved — DELETE
// returns NotFound and drops nothing, instead of silently destroying an
// unrelated table and reporting a bare 404.
//
// Load-bearing: remove the TableHasPlatformSystemColumns corroboration from
// shouldReconcileOrphanTable and this fails — the reconcile drops the raw table.
// TestDeleteSchema_ReconcileRefusesUnregisteredTable is the data-loss guard the
// review demanded: an absent-metadata DELETE for a name whose table has NO
// _System_Table registry row — a raw operator/ETL/backup table, INCLUDING a
// same-shape CREATE TABLE ... LIKE copy that defeated the earlier shape
// heuristic — must stay a fail-closed 404 no-op (base behavior) and never drop.
func TestDeleteSchema_ReconcileRefusesUnregisteredTable(t *testing.T) {
	sr := &reconcileSchemaRepo{notOwned: true}
	ms := NewMetadataService(&reconcileMetaRepo{}, NewSchemaManager(sr), nil)

	for _, name := range []string{"operator_backup_2026", "Widget_backup"} {
		err := ms.DeleteSchema(context.Background(), name)
		if !apperrors.IsNotFound(err) {
			t.Fatalf("DeleteSchema(%q) error = %T %v, want NotFound", name, err, err)
		}
	}
	if len(sr.dropped) != 0 {
		t.Fatalf("DropTable calls = %v, want none (a table without a registry tombstone must never be dropped)", sr.dropped)
	}
}

// TestDeleteSchema_ReconcileRefusesOnRegistryProbeError verifies the tombstone
// probe fails CLOSED: if the _System_Table lookup itself errors, the reconcile
// must refuse the destructive drop rather than proceed on unknown provenance.
func TestDeleteSchema_ReconcileRefusesOnRegistryProbeError(t *testing.T) {
	sr := &reconcileSchemaRepo{probeErr: errors.New("registry probe failed")}
	ms := NewMetadataService(&reconcileMetaRepo{}, NewSchemaManager(sr), nil)

	err := ms.DeleteSchema(context.Background(), "Widget__c")
	if !apperrors.IsNotFound(err) {
		t.Fatalf("DeleteSchema error = %T %v, want NotFound", err, err)
	}
	if len(sr.dropped) != 0 {
		t.Fatalf("DropTable calls = %v, want none (a registry-probe error must fail closed)", sr.dropped)
	}
}