package records

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"net"
	"os"
	"path/filepath"
	"testing"
	"time"

	"github.com/go-sql-driver/mysql"
	"github.com/nexuscrm/kernel/pkg/schemacatalog"
	metadataFeature "github.com/nexuscrm/metadata"
	"github.com/nexuscrm/permissions"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/contract"
	sharedErrors "github.com/nexuscrm/shared/pkg/errors"
	"github.com/nexuscrm/shared/pkg/models"
	permissionsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/effectivepermissionsv1"
	defaultsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recorddefaultsv1"
)

// This fixture supplies only authenticated system capability facts. Actual
// profile/assignment/FLS aggregation and missing-row behavior run on TiDB.
type fieldProjectionPermissionRepo struct {
	*persistence.PermissionRepository
	system map[string]bool
}

func (r *fieldProjectionPermissionRepo) GetEffectiveSystemPermissions(context.Context, string, string) (map[string]bool, error) {
	return r.system, nil
}

type fieldProjectionEvidence struct {
	Case     string
	Object   models.ObjectPermissions
	Fields   []contract.FieldAccessProjection
	Defaults []defaultsemantic.Field
}

// The independent oracle is the existing absence-vs-explicit-denial contract,
// metadata hard restrictions, and distinct insert/update capabilities. It is
// checked at point, cached point, complete/bounded projections, and Records'
// default producer, so returning stored FLS rows as decisions cannot pass.
func TestEffectiveFieldPermissionProjectionTiDB(t *testing.T) {
	db := openFieldProjectionTiDB(t)
	ctx, cancel := context.WithTimeout(t.Context(), 120*time.Second)
	defer cancel()
	definitions, err := schemacatalog.SystemTableDefinitions()
	if err != nil {
		t.Fatal(err)
	}
	required := map[string]bool{"PermissionSet": true, "PermissionSetAssignment": true, "PermissionSetGroup": true, "PermissionSetGroupComponent": true, "PermissionSetGroupAssignment": true, "ObjectPermissions": true, "FieldPermissions": true, "UserRole": true}
	schemaRepo := persistence.NewSchemaRepository(db, nil)
	for _, definition := range definitions {
		if required[definition.TableName] {
			if err := schemaRepo.CreatePhysicalTable(ctx, definition); err != nil {
				t.Fatal(err)
			}
			delete(required, definition.TableName)
		}
	}
	if len(required) != 0 {
		t.Fatalf("missing canonical definitions: %v", required)
	}
	exec := func(query string, args ...any) {
		t.Helper()
		if _, err := db.ExecContext(ctx, query, args...); err != nil {
			t.Fatal(err)
		}
	}
	exec(`INSERT INTO PermissionSet (Id,Name,Label,IsDeleted,IsOwnedByProfile) VALUES ('field-set','FieldSet','Field Set',0,0)`)
	exec(`INSERT INTO PermissionSetAssignment (Id,AssigneeId,PermissionSetId,IsDeleted) VALUES ('field-assignment','field-user','field-set',0)`)
	exec(`INSERT INTO ObjectPermissions (Id,ParentId,SobjectType,PermissionsRead,PermissionsCreate,PermissionsEdit,IsDeleted) VALUES ('field-object','field-set','Entry',1,1,1,0)`)
	// Explicit FLS cannot make audit fields editable.
	exec(`INSERT INTO FieldPermissions (Id,ParentId,SobjectType,Field,PermissionsRead,PermissionsEdit,IsDeleted) VALUES ('field-audit','field-set','Entry','Entry.Id',1,1,0)`)
	exec(`INSERT INTO PermissionSetGroup (Id,ApiName,Label,IsActive,IsDeleted) VALUES ('field-group','FieldGroup','Field Group',1,0)`)
	exec(`INSERT INTO PermissionSetGroupComponent (Id,GroupId,PermissionSetId) VALUES ('field-component','field-group','field-set')`)
	exec(`INSERT INTO PermissionSetGroupAssignment (Id,AssigneeId,GroupId,IsDeleted) VALUES ('field-group-assignment','field-user','field-group',1)`)
	var evidence []fieldProjectionEvidence
	for _, source := range []string{"direct", "profile", "group"} {
		t.Run(source, func(t *testing.T) {
			exec(`UPDATE PermissionSet SET IsOwnedByProfile=?,ProfileId=? WHERE Id='field-set'`, source == "profile", "field-profile")
			exec(`UPDATE PermissionSetAssignment SET IsDeleted=? WHERE Id='field-assignment'`, source != "direct")
			exec(`UPDATE PermissionSetGroupAssignment SET IsDeleted=? WHERE Id='field-group-assignment'`, source != "group")
			for _, tc := range []struct {
				name                                          string
				objectRead, objectCreate, objectEdit          bool
				explicit, read, edit, blocked, super, derived bool
				wantRead, wantCreate, wantEdit                bool
			}{
				{name: "missing", objectRead: true, objectCreate: true, objectEdit: true, wantRead: true, wantCreate: true, wantEdit: true},
				{name: "create_only", objectRead: true, objectCreate: true, wantRead: true, wantCreate: true},
				{name: "edit_only", objectRead: true, objectEdit: true, wantRead: true, wantEdit: true},
				{name: "read_only", objectRead: true, wantRead: true},
				{name: "no_object_access"},
				{name: "explicit_allow", objectRead: true, objectCreate: true, objectEdit: true, explicit: true, read: true, edit: true, wantRead: true, wantCreate: true, wantEdit: true},
				{name: "explicit_deny", objectRead: true, objectCreate: true, objectEdit: true, explicit: true},
				{name: "explicit_read_only", objectRead: true, objectCreate: true, objectEdit: true, explicit: true, read: true, wantRead: true},
				{name: "blocked_allow", objectRead: true, objectCreate: true, objectEdit: true, explicit: true, read: true, edit: true, blocked: true, wantCreate: true, wantEdit: true},
				{name: "blocked_missing", objectRead: true, objectCreate: true, objectEdit: true, blocked: true, wantCreate: true, wantEdit: true},
				{name: "super_missing", super: true, wantRead: true, wantCreate: true, wantEdit: true},
				{name: "super_blocked", super: true, blocked: true, wantCreate: true, wantEdit: true},
				{name: "derived_create_not_update", derived: true, objectRead: true, objectCreate: true, objectEdit: true, wantCreate: true},
			} {
				t.Run(tc.name, func(t *testing.T) {
					exec(`UPDATE ObjectPermissions SET PermissionsRead=?,PermissionsCreate=?,PermissionsEdit=? WHERE Id='field-object'`, tc.objectRead, tc.objectCreate, tc.objectEdit)
					exec(`DELETE FROM FieldPermissions WHERE Id='field-owner'`)
					if tc.explicit {
						exec(`INSERT INTO FieldPermissions (Id,ParentId,SobjectType,Field,PermissionsRead,PermissionsEdit,IsDeleted) VALUES ('field-owner','field-set','Entry','Entry.OwnerId',?,?,0)`, tc.read, tc.edit)
					}
					schema := &models.ObjectMetadata{APIName: "Entry", SharingModel: constants.SharingModelPrivate, Fields: []models.FieldMetadata{
						{APIName: "Id", Type: constants.FieldTypeString},
						{APIName: "Name", Type: constants.FieldTypeString, Createable: true, Updateable: true},
						{APIName: "OwnerId", Type: constants.FieldTypeReference, ReferenceTo: []string{"User"}, Createable: true, Updateable: true},
					}}
					if tc.blocked {
						schema.Fields[2].ReadProtectionPolicy = constants.FieldReadProtectionBlocked
					}
					if tc.derived {
						schema.AccessPolicy = &models.ObjectAccessPolicy{Mode: "derived",
							ObjectRead:   &models.AccessPolicyExpr{Kind: "systemPermission", APIName: "FixtureRead"},
							ObjectCreate: &models.AccessPolicyExpr{Kind: "systemPermission", APIName: "FixtureCreate"},
							ObjectEdit:   &models.AccessPolicyExpr{Kind: "systemPermission", APIName: "FixtureEdit"},
							FieldRead:    &models.AccessPolicyExpr{Kind: "systemPermission", APIName: "FixtureFieldRead"},
							FieldEdit:    &models.AccessPolicyExpr{Kind: "systemPermission", APIName: "FixtureFieldEdit"},
						}
					}
					makeFieldsMap(schema)
					other := &models.ObjectMetadata{APIName: "OtherEntry", Fields: []models.FieldMetadata{{APIName: "Name", Type: constants.FieldTypeString}}}
					makeFieldsMap(other)
					meta := metadataFeature.NewMetadataService(&stubUpsertMetaRepo{schemas: []*models.ObjectMetadata{schema, other}}, nil, nil)
					repo := &fieldProjectionPermissionRepo{PermissionRepository: persistence.NewPermissionRepository(db), system: map[string]bool{"ModifyAllData": tc.super, "FixtureRead": true, "FixtureCreate": true, "FixtureEdit": true}}
					service := permissions.NewPermissionService(repo, meta, nil)
					user := &models.UserSession{ID: "field-user", ProfileID: "field-profile", Name: "Field User"}
					owner := models.FindField(schema, "OwnerId")
					resolver := permissions.NewFieldAccessResolver(service)
					for _, cached := range []bool{false, true} {
						checkCtx := ctx
						if cached {
							checkCtx = permissions.WithAccessInputCache(checkCtx)
						}
						read, err := service.CheckFieldVisibilityWithSchema(checkCtx, schema, owner, user)
						if err != nil {
							t.Fatal(err)
						}
						edit, err := service.CheckFieldEditabilityWithUser(checkCtx, "Entry", "OwnerId", user)
						if err != nil {
							t.Fatal(err)
						}
						create, err := resolver.EffectiveCreateable(checkCtx, owner, "Entry", user)
						if err != nil {
							t.Fatal(err)
						}
						if read != tc.wantRead || edit != tc.wantEdit || create != tc.wantCreate {
							t.Fatalf("cached=%v point read/create/edit=%v/%v/%v; want %v/%v/%v", cached, read, create, edit, tc.wantRead, tc.wantCreate, tc.wantEdit)
						}
					}
					for _, scope := range []string{"all", "one", "many"} {
						var rows []models.FieldPermissions
						var err error
						switch scope {
						case "all":
							rows, err = service.ListEffectiveFieldPermissionsForUser(ctx, user.ID, user.ProfileID)
						case "one":
							rows, err = service.ListEffectiveFieldPermissionsForUserObject(ctx, user.ID, user.ProfileID, "Entry")
						case "many":
							rows, err = service.ListEffectiveFieldPermissionsForUserObjects(ctx, user.ID, user.ProfileID, []string{"Entry", "OtherEntry"})
						}
						if err != nil {
							t.Fatal(err)
						}
						found := false
						for _, row := range rows {
							if scope == "one" && row.SobjectType != "Entry" {
								t.Fatalf("bounded projection leaked %s", row.SobjectType)
							}
							if row.SobjectType == "Entry" && row.Field == "OwnerId" {
								found = true
								if row.PermissionsRead != tc.wantRead || row.PermissionsEdit != tc.wantEdit {
									t.Fatalf("%s projection owner=%+v", scope, row)
								}
							}
							if row.Field == "Id" && row.PermissionsEdit {
								t.Fatalf("%s made audit field editable", scope)
							}
						}
						if !found {
							t.Fatalf("%s omitted the resolved OwnerId decision", scope)
						}
					}
					projected, err := permissions.NewContractPermissionAdmin(service, nil).EffectivePermissionsSelected(ctx, user, "", permissionsemantic.Selection{Sections: []permissionsemantic.Section{permissionsemantic.SectionFieldPermissions}, ResourceTypes: []string{"Entry"}})
					if err != nil {
						t.Fatal(err)
					}
					for _, field := range projected.FieldPermissions {
						if field.FieldName == "OwnerId" && (field.CanRead != tc.wantRead || field.CanEdit != tc.wantEdit || field.CanCreate != tc.wantCreate) {
							t.Fatalf("contract projection=%+v", field)
						}
					}
					defaults, err := (&PersistenceService{metadata: meta, permissions: service}).PreviewCreateDefaults(ctx, defaultsemantic.Request{ObjectID: "Entry"}, user)
					if !(tc.objectRead && tc.objectCreate) && !tc.super {
						if !sharedErrors.IsPermissionDenied(err) {
							t.Fatalf("default preview should deny object access, got %v", err)
						}
					} else if err != nil {
						t.Fatal(err)
					}
					if (len(defaults) > 0) != (tc.wantRead && tc.wantCreate) {
						t.Fatalf("defaults=%+v do not match read/create authority", defaults)
					}
					effectiveSchema, err := service.GetEffectiveSchema(ctx, schema, user)
					if err != nil {
						t.Fatal(err)
					}
					if len(schema.Fields) != 3 || models.FindField(schema, "OwnerId") == nil {
						t.Fatal("source metadata was mutated")
					}
					if (models.FindField(effectiveSchema, "OwnerId") != nil) != tc.wantRead {
						t.Fatal("effective schema disagrees with read restriction")
					}
					evidence = append(evidence, fieldProjectionEvidence{Case: source + "/" + tc.name, Object: models.ObjectPermissions{SobjectType: "Entry", PermissionsRead: tc.objectRead || tc.super, PermissionsCreate: tc.objectCreate || tc.super, PermissionsEdit: tc.objectEdit || tc.super}, Fields: projected.FieldPermissions, Defaults: defaults})
				})
			}
		})
	}
	if path := os.Getenv("NEXUSCRM_FIELD_PERMISSION_EVIDENCE"); path != "" {
		data, err := json.MarshalIndent(evidence, "", "  ")
		if err != nil {
			t.Fatal(err)
		}
		if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil {
			t.Fatal(err)
		}
		if err := os.WriteFile(path, data, 0600); err != nil {
			t.Fatal(err)
		}
	}
}

func openFieldProjectionTiDB(t *testing.T) *sql.DB {
	t.Helper()
	host := os.Getenv("TIDB_HOST")
	if host == "" {
		if os.Getenv("NEXUSCRM_TIDB_REQUIRED") == "1" {
			t.Fatal("TIDB_HOST is required for field permission regression")
		}
		t.Skip("field permission integration requires TIDB_HOST")
	}
	port := os.Getenv("TIDB_PORT")
	if port == "" {
		port = "4000"
	}
	user := os.Getenv("TIDB_USER")
	if user == "" {
		user = "root"
	}
	cfg := mysql.NewConfig()
	cfg.Net = "tcp"
	cfg.Addr = net.JoinHostPort(host, port)
	cfg.User = user
	cfg.Passwd = os.Getenv("TIDB_PASSWORD")
	cfg.ParseTime = true
	cfg.Timeout = 5 * time.Second
	cfg.ReadTimeout = 10 * time.Second
	cfg.WriteTimeout = 10 * time.Second
	cfg.Collation = "utf8mb4_unicode_ci"
	admin, err := sql.Open("mysql", cfg.FormatDSN())
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { admin.Close() })
	if err := admin.PingContext(t.Context()); err != nil {
		t.Fatal(err)
	}
	name := fmt.Sprintf("field_projection_%d", time.Now().UnixNano())
	if _, err := admin.ExecContext(t.Context(), "CREATE DATABASE `"+name+"`"); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() {
		if _, err := admin.Exec("DROP DATABASE `" + name + "`"); err != nil {
			t.Error(err)
		}
	})
	cfg.DBName = name
	db, err := sql.Open("mysql", cfg.FormatDSN())
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { db.Close() })
	return db
}