//go:build realconv

// real_chain_e2e_test.go — 完整鏈路 e2e：真轉檔 → PromoteToModels → 真 model 進 PG 模型庫。
//
// Owner: testing agent（realconv 完整鏈路 e2e — build tag 隔離，預設 CI 不跑）
//
// 與 real_converter_e2e_test.go 的差異：
//   - real_converter_e2e_test.go：對「真轉檔服務」驗 visionA ConverterClient 的 **contract**
//     （連線/認證/InitJob/GetJob/GetResult 解析、promote 失敗路徑）。當時 stage 是 stub。
//   - 本檔：stage 已切 **real 模式**（真 KTC、真 nef 800KB、promote OAuth 已修），驗
//     **整條 visionA 業務鏈路到底**：
//     InitJob（真送 onnx+56 圖）→ poll completed（真轉檔）
//     → flow.PromoteToModels（promote→converter MinIO pull NEF→storage.Put→建 model record）
//     → **model 真的進了 PG 模型庫**（用 model.PostgresRepository 查、驗 owner/source_job_id/storage_key/faa_object_key）
//     → 冪等（同 jobID 再 promote 回既有 model、不重複建）
//
// ⚠️ v0.6 架構事實（flow.go PromoteToModels）：promote 與 download 都走
//
//	`converter.GetResult`（converter MinIO），visionA 端**不再直接打 FAA**。所以本鏈路
//	**不需要** wire FAA / MC client —— 只需要：真 ConverterClient（:9501 + API key）
//	+ 真 PG（建 model record）+ 一個 storage（streaming 寫 NEF；用本機 tmpdir LocalFS）。
//	這是「能在本檔組出真 conversion service」的關鍵：依賴比想像中少。
//
// 三個必要外部資源（缺任一 → t.Skip 並印啟用指令）：
//  1. VISIONA_REAL_CONVERTER_URL + VISIONA_CONVERTER_API_KEY（與 real_converter_e2e_test.go 共用）
//  2. VISIONA_REAL_PG_DSN（已 migrate 的 PG；可指 stage 真 PG 或本機 docker PG）
//  3. onnx + ref images fixture（與 real_converter_e2e_test.go 共用 default 路徑 / env 覆寫）
//
// 本機通常連不到 stage（VPN/網路）＋ 無 PG → 預設 Skip。由 Orchestrator 對 stage 跑
// （見檔尾「給 Orchestrator 對 stage 跑」段）。
package conversion

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"os"
	"testing"
	"time"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5/pgxpool"

	"visiona-backend/internal/model"
	"visiona-backend/internal/storage"
)

// ==========================================================================
// 環境 guard（PG DSN 是本檔額外需要的；converter / fixture 沿用 requireRealConvEnv）
// ==========================================================================

const (
	// realChainPGDSNEnv 是已 migrate 的 PostgreSQL DSN（如
	// postgres://visiona:pw@192.168.0.130:5432/visiona?sslmode=disable）。
	// 缺 → 本檔的完整鏈路測試 Skip（contract 測試在 real_converter_e2e_test.go 仍可獨立跑）。
	realChainPGDSNEnv = "VISIONA_REAL_PG_DSN"

	// 真 KTC 轉檔比 stub 慢得多（onnx→bie→nef 三 stage、bie 量化要跑 56 張 ref 圖）。
	// 給足 deadline；stub 經驗幾秒，real 端視機器可能數分鐘。
	realChainPollTimeout  = 10 * time.Minute
	realChainPollInterval = 3 * time.Second

	// 測試建立的 model name 前綴（可識別、cleanup 用）。
	realChainModelNamePrefix = "e2e-realchain-"
)

// requireRealChainPG 解析 PG DSN、建 pool；缺 env → Skip。pool 在 t.Cleanup 關閉。
func requireRealChainPG(t *testing.T) *pgxpool.Pool {
	t.Helper()
	dsn := os.Getenv(realChainPGDSNEnv)
	if dsn == "" {
		t.Skipf(`real-chain e2e 跳過：未設 %s（PG DSN）。
本測試需要一個「已 migrate（含 users / models 表）」的 PostgreSQL：
  - 指 stage 真 PG，或本機 docker PG（schema 來自 migrations/0001_create_users_models.up.sql）
  範例：
    VISIONA_REAL_PG_DSN="postgres://visiona:<pw>@192.168.0.130:5432/visiona?sslmode=disable"`, realChainPGDSNEnv)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
	defer cancel()
	pool, err := pgxpool.New(ctx, dsn)
	if err != nil {
		t.Skipf("real-chain e2e 跳過：PG pool 建立失敗（DSN 可達性問題？）：%v", err)
	}
	// ping 確認連得上 + schema 存在（查 models 表）
	pingCtx, pingCancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer pingCancel()
	if err := pool.Ping(pingCtx); err != nil {
		pool.Close()
		t.Skipf("real-chain e2e 跳過：PG ping 失敗：%v", err)
	}
	var reg int
	if err := pool.QueryRow(pingCtx, `SELECT 1 FROM information_schema.tables
		WHERE table_name = 'models'`).Scan(&reg); err != nil {
		pool.Close()
		t.Skipf("real-chain e2e 跳過：PG 缺 models 表（DSN 指到未 migrate 的 DB？）：%v。"+
			"請先對該 DB 跑 migrations/0001_create_users_models.up.sql。", err)
	}

	t.Cleanup(func() { pool.Close() })
	return pool
}

// ensureTestUser upsert 一個合法 UUID user（滿足 models.owner_user_id FK）。
//
// 回傳該 user 的 UUID 字串。固定 UUID（deterministic）讓重跑時 idempotent；
// email 帶可識別前綴避免撞真 user。
func ensureTestUser(t *testing.T, pool *pgxpool.Pool) string {
	t.Helper()
	// 固定 namespace UUID（v5 不需要；這裡直接寫死一個明顯是測試用的 UUID）。
	const testUserID = "e2e0c0de-0000-4000-8000-000000000001"
	const testEmail = "e2e-realchain@example.invalid"

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	// upsert by id；email 用 functional unique index（lower(email)），ON CONFLICT (id) 即可。
	_, err := pool.Exec(ctx, `
		INSERT INTO users (id, email, name, roles)
		VALUES ($1, $2, 'e2e realchain test user', '{}')
		ON CONFLICT (id) DO UPDATE SET email = EXCLUDED.email, updated_at = now()`,
		testUserID, testEmail)
	if err != nil {
		t.Fatalf("ensure test user（upsert users）失敗：%v\n"+
			"（若是 email unique 衝突，代表有殘留同 email 的別的 user — 換 testEmail 或先清）", err)
	}
	return testUserID
}

// ==========================================================================
// in-test adapters：把 model.PostgresRepository / storage.LocalFSStore 包成
// conversion.ModelStore / conversion.Storage（對映 cmd/api-server/conversion_adapters.go，
// 但 main package 的 adapter 不可 import，這裡在 conversion package 內等價重寫）。
// ==========================================================================

// pgModelStore 把 model.Repository 包成 conversion.ModelStore（含 ModelRecord ↔ model.Model 轉換）。
type pgModelStore struct {
	repo model.Repository
}

func (s *pgModelStore) Save(ctx context.Context, rec *ModelRecord) error {
	if rec == nil {
		return errors.New("pgModelStore.Save requires non-nil record")
	}
	now := time.Now().UTC()
	uploadedAt := now
	if !rec.UpdatedAt.IsZero() {
		uploadedAt = rec.UpdatedAt
	}
	m := &model.Model{
		ID:           rec.ID,
		OwnerUserID:  rec.OwnerUserID,
		Name:         rec.Name,
		Description:  rec.Description,
		StorageKey:   rec.StorageKey,
		FileSize:     rec.FileSize,
		FileChecksum: rec.FileChecksum,
		TargetChip:   rec.TargetChip,
		InputShape:   rec.InputShape,
		Classes:      rec.Classes,
		Framework:    rec.Framework,
		Source:       rec.Source,
		SourceJobID:  rec.SourceJobID,
		FAAObjectKey: rec.FAAObjectKey,
		CreatedAt:    rec.CreatedAt,
		UpdatedAt:    rec.UpdatedAt,
		UploadedAt:   &uploadedAt,
	}
	return s.repo.Save(ctx, m)
}

func (s *pgModelStore) FindBySourceJobID(ctx context.Context, ownerUserID, sourceJobID string) (*ModelRecord, error) {
	if ownerUserID == "" || sourceJobID == "" {
		return nil, nil
	}
	models, err := s.repo.List(ctx, model.ListFilter{
		OwnerUserID: ownerUserID,
		Source:      model.SourceConverted,
	})
	if err != nil {
		return nil, fmt.Errorf("pgModelStore.FindBySourceJobID list: %w", err)
	}
	for _, m := range models {
		if m.SourceJobID == sourceJobID {
			return &ModelRecord{
				ID:           m.ID,
				OwnerUserID:  m.OwnerUserID,
				Name:         m.Name,
				Description:  m.Description,
				StorageKey:   m.StorageKey,
				FileSize:     m.FileSize,
				FileChecksum: m.FileChecksum,
				TargetChip:   m.TargetChip,
				InputShape:   m.InputShape,
				Classes:      m.Classes,
				Framework:    m.Framework,
				Source:       m.Source,
				SourceJobID:  m.SourceJobID,
				FAAObjectKey: m.FAAObjectKey,
				CreatedAt:    m.CreatedAt,
				UpdatedAt:    m.UpdatedAt,
			}, nil
		}
	}
	return nil, nil
}

func (s *pgModelStore) GenerateID() string { return uuid.NewString() }

// localStorage 把 storage.Store 包成 conversion.Storage（只需 Put）。
type localStorage struct {
	store storage.Store
}

func (s *localStorage) Put(ctx context.Context, key string, r io.Reader, size int64, meta map[string]string) error {
	return s.store.Put(ctx, key, r, size, meta)
}

// buildRealChainService 組一個「真 converter + 真 PG + 本機 LocalFS storage」的 conversion.Service。
//
// 回傳 service + 底層 pgRepo（測試直接用 repo 查 PG 驗證 model 落盤）。
func buildRealChainService(t *testing.T, env realConvEnv, pool *pgxpool.Pool) (Service, *model.PostgresRepository) {
	t.Helper()
	logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelWarn}))

	converterClient := NewConverterClient(ConverterClientOpts{
		BaseURL: env.baseURL,
		APIKey:  env.apiKey,
		Logger:  logger,
	})
	ownership := NewOwnership(converterClient, logger)

	pgRepo := model.NewPostgresRepository(pool)
	modelStore := &pgModelStore{repo: pgRepo}

	// 本機 tmpdir LocalFS storage（NEF streaming 寫進去；測完隨 t.TempDir 清）。
	fsStore, err := storage.NewLocalFSStore(t.TempDir(), "http://localhost/files", "e2e-realchain-signing")
	if err != nil {
		t.Fatalf("建 LocalFSStore 失敗：%v", err)
	}
	storageAdapter := &localStorage{store: fsStore}

	svc, err := NewService(FlowOpts{
		Converter:  converterClient,
		Ownership:  ownership,
		ModelStore: modelStore,
		Storage:    storageAdapter,
		Logger:     logger,
	})
	if err != nil {
		t.Fatalf("NewService 失敗：%v", err)
	}
	return svc, pgRepo
}

// ==========================================================================
// E2E：完整鏈路 — 真轉檔 → PromoteToModels → 真 model 進 PG → 冪等
// ==========================================================================

// TestRealChain_ConvertPromoteToPGModelLibrary 驗整條 visionA 業務鏈路到底。
//
//	[1] flow.InitJob 真送 onnx + 56 圖（flow 內部重組 multipart、注入 user_id、寫 ownership）
//	[2] flow.GetJob poll 到 completed（真 KTC 轉檔；給足 10 分鐘）
//	[3] flow.PromoteToModels：promote（真 nef 推上 FAA + 保留 converter MinIO）
//	    → converter.GetResult 拉 NEF stream → storage.Put → model.PostgresRepository.Save
//	[4] 驗 model 真的進 PG：用 pgRepo.Get(model_id) + List by owner，確認
//	    owner / source_job_id / storage_key / file_size / source=converted / faa_object_key 有值
//	[5] 冪等：同 jobID 再 PromoteToModels → 回**同一個** model_id（不重複建）
func TestRealChain_ConvertPromoteToPGModelLibrary(t *testing.T) {
	env := requireRealConvEnv(t)
	pool := requireRealChainPG(t)
	userID := ensureTestUser(t, pool)
	svc, pgRepo := buildRealChainService(t, env, pool)

	ctx := context.Background()

	// ── [1] flow.InitJob（真送）─────────────────────────────────────────────
	body, contentType, refCount := buildRealInitBody(t, env)
	t.Logf("[1] InitJob：multipart %d bytes，ref_images=%d 張，user=%s", len(body), refCount, userID)

	// 注意：flow.InitJob 內部會重組 multipart 並注入 user_id（黑名單 client 帶來的 user_id）。
	// buildRealInitBody 已寫了 user_id=realConvTestUserID，但會被 flow 用本測 userID 蓋掉，
	// 這正是要驗的安全行為（§4.2）。傳進去的 ContentType 必須含 boundary。
	initCtx, initCancel := context.WithTimeout(ctx, 90*time.Second)
	job, err := svc.InitJob(initCtx, InitJobInput{
		UserID:        userID,
		ContentType:   contentType,
		Body:          bytes.NewReader(body),
		ContentLength: int64(len(body)),
	})
	initCancel()
	if err != nil {
		if errors.Is(err, ErrConverterAuthFailed) {
			t.Fatalf("[1] InitJob 認證失敗（API key 未對齊？）：%v", err)
		}
		t.Fatalf("[1] InitJob 失敗：%v", err)
	}
	if job.JobID == "" {
		t.Fatalf("[1] InitJob 回的 job_id 為空：%+v", job)
	}
	jobID := job.JobID
	t.Logf("[1] InitJob OK：job_id=%s status=%s stage=%s", jobID, job.Status, job.Stage)

	// ── [2] flow.GetJob poll 到 completed（真轉檔，給足 10 分鐘）──────────────
	final := pollChainUntilTerminal(t, svc, userID, jobID)
	t.Logf("[2] 終態：status=%s stage=%q source_filename=%q target_chip=%q error_code=%q",
		final.Status, final.Stage, final.SourceFilename, final.TargetChip, final.ErrorCode)
	if final.Status != "completed" {
		t.Fatalf("[2] 真轉檔未 completed（status=%s error_code=%q msg=%q）。"+
			"確認 stage worker 已切 real 模式、且該 fixture 能轉成功。",
			final.Status, final.ErrorCode, final.ErrorMessage)
	}

	// ── [3] flow.PromoteToModels（真 promote → MinIO pull → storage → 建 model record）──
	modelName := realChainModelNamePrefix + jobID[:8]
	promoteCtx, promoteCancel := context.WithTimeout(ctx, 2*time.Minute)
	promoteRes, err := svc.PromoteToModels(promoteCtx, userID, jobID, modelName)
	promoteCancel()
	if err != nil {
		t.Fatalf("[3] PromoteToModels 失敗：%v\n"+
			"（promote OAuth 已修的前提下不該失敗；若回 ErrConverterUnavailable 代表 promote→FAA 仍有問題，"+
			"回報 Orchestrator，勿自行改 production code。）", err)
	}
	if promoteRes == nil || promoteRes.ModelID == "" {
		t.Fatalf("[3] PromoteToModels 回的 model_id 為空：%+v", promoteRes)
	}
	modelID := promoteRes.ModelID
	// cleanup：測完軟刪除 model（pgRepo.Delete 寫 deleted_at；不留垃圾在模型庫）。
	t.Cleanup(func() {
		dctx, dcancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer dcancel()
		if derr := pgRepo.Delete(dctx, modelID); derr != nil {
			t.Logf("cleanup：軟刪除 model %s 失敗（殘留在 PG，需人工清）：%v", modelID, derr)
		} else {
			t.Logf("cleanup：已軟刪除 model %s", modelID)
		}
	})
	t.Logf("[3] PromoteToModels OK：model_id=%s name=%q source=%s source_job_id=%s file_size=%d status=%s",
		modelID, promoteRes.Name, promoteRes.Source, promoteRes.SourceJobID, promoteRes.FileSize, promoteRes.Status)

	// 基本一致性
	if promoteRes.Source != "converted" {
		t.Errorf("[3] promote source 預期 converted，得 %q", promoteRes.Source)
	}
	if promoteRes.SourceJobID != jobID {
		t.Errorf("[3] promote source_job_id 預期 %s，得 %s", jobID, promoteRes.SourceJobID)
	}
	if promoteRes.FileSize <= 0 {
		t.Errorf("[3] promote file_size 應 > 0（真 nef ~800KB），得 %d", promoteRes.FileSize)
	}

	// ── [4] 驗 model 真的進 PG（直接查 PostgresRepository）────────────────────
	getCtx, getCancel := context.WithTimeout(ctx, 10*time.Second)
	got, err := pgRepo.Get(getCtx, modelID)
	getCancel()
	if err != nil {
		t.Fatalf("[4] 從 PG 查不到 model %s（model 未真正落盤？）：%v", modelID, err)
	}
	t.Logf("[4] PG model 落盤：id=%s owner=%s name=%q storage_key=%q faa_object_key=%q "+
		"file_size=%d source=%s source_job_id=%s target_chip=%q",
		got.ID, got.OwnerUserID, got.Name, got.StorageKey, got.FAAObjectKey,
		got.FileSize, got.Source, got.SourceJobID, got.TargetChip)

	if got.OwnerUserID != userID {
		t.Errorf("[4] PG model owner 預期 %s，得 %s", userID, got.OwnerUserID)
	}
	if got.SourceJobID != jobID {
		t.Errorf("[4] PG model source_job_id 預期 %s，得 %s", jobID, got.SourceJobID)
	}
	if got.Source != model.SourceConverted {
		t.Errorf("[4] PG model source 預期 converted，得 %q", got.Source)
	}
	if got.StorageKey == "" {
		t.Errorf("[4] PG model storage_key 不該為空（promote 應寫 visionA storage key）")
	}
	if got.FileSize <= 0 {
		t.Errorf("[4] PG model file_size 應 > 0，得 %d", got.FileSize)
	}
	// FAAObjectKey：v0.6 promote 仍寫此欄位（= converter promote 的 target_object_key）。
	// 真 promote 成功路徑下應有值；若為空記下供判讀（不一定 fail — 視 converter promote response）。
	if got.FAAObjectKey == "" {
		t.Logf("[4] 注意：PG model faa_object_key 為空。v0.6 promote 應回 target_object_key；" +
			"若 converter promote response 未帶 target_object_key 則 fallback 為 visionA 端組的 key，" +
			"理論上不該空。確認 converter promote 回傳格式。")
	}

	// List by owner 也應看得到（驗 List 路徑 + filter 正確）
	listCtx, listCancel := context.WithTimeout(ctx, 10*time.Second)
	models, err := pgRepo.List(listCtx, model.ListFilter{OwnerUserID: userID, Source: model.SourceConverted})
	listCancel()
	if err != nil {
		t.Fatalf("[4] List by owner 失敗：%v", err)
	}
	if !containsModelID(models, modelID) {
		t.Errorf("[4] List by owner=%s source=converted 結果未含 model %s（共 %d 筆）",
			userID, modelID, len(models))
	}

	// ── [5] 冪等：同 jobID 再 PromoteToModels → 回同一個 model_id ──────────────
	idemCtx, idemCancel := context.WithTimeout(ctx, 1*time.Minute)
	promoteRes2, err := svc.PromoteToModels(idemCtx, userID, jobID, modelName)
	idemCancel()
	if err != nil {
		t.Fatalf("[5] 第二次 PromoteToModels（冪等）失敗：%v", err)
	}
	if promoteRes2 == nil || promoteRes2.ModelID != modelID {
		t.Errorf("[5] 冪等失敗：第二次 promote 應回同一 model_id=%s，得 %+v", modelID, promoteRes2)
	} else {
		t.Logf("[5] 冪等 OK：同 jobID 再 promote 回既有 model_id=%s（未重複建）", modelID)
	}

	// 再查一次 PG 確認只有一筆 converted model 對應此 jobID（冪等不該建第二筆）
	cntCtx, cntCancel := context.WithTimeout(ctx, 10*time.Second)
	all, err := pgRepo.List(cntCtx, model.ListFilter{OwnerUserID: userID, Source: model.SourceConverted})
	cntCancel()
	if err != nil {
		t.Fatalf("[5] 冪等後 List 失敗：%v", err)
	}
	n := 0
	for _, m := range all {
		if m.SourceJobID == jobID {
			n++
		}
	}
	if n != 1 {
		t.Errorf("[5] 冪等後對應 jobID=%s 的 converted model 應只有 1 筆，得 %d 筆", jobID, n)
	}

	t.Logf("完整鏈路驗證通過：真轉檔 → PromoteToModels → model 進 PG → 冪等，全部 OK。")
}

// pollChainUntilTerminal 用 flow.GetJob 對真服務 poll 到 completed/failed 或 timeout。
//
// 與 real_converter_e2e_test.go 的 pollUntilTerminal 不同：這裡走 **flow.GetJob**（含 ownership
// 檢查），驗的是 visionA 業務層的 poll，而非 raw client。回傳對外 *Job。
func pollChainUntilTerminal(t *testing.T, svc Service, userID, jobID string) *Job {
	t.Helper()
	deadline := time.Now().Add(realChainPollTimeout)
	var last *Job
	logEvery := 10 // 每 N 次 poll 印一次進度，避免長轉檔時 log 太吵
	i := 0
	for time.Now().Before(deadline) {
		ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
		j, err := svc.GetJob(ctx, userID, jobID)
		cancel()
		if err != nil {
			t.Logf("GetJob 暫時失敗（續 poll）：%v", err)
			time.Sleep(realChainPollInterval)
			continue
		}
		last = j
		i++
		if i%logEvery == 0 {
			t.Logf("  poll #%d：status=%s stage=%q progress=%d stage_progress=%d",
				i, j.Status, j.Stage, j.Progress, j.StageProgress)
		}
		switch j.Status {
		case "completed", "failed":
			return j
		default:
			time.Sleep(realChainPollInterval)
		}
	}
	if last == nil {
		t.Fatalf("poll job %s 超時且從未成功 GetJob", jobID)
	}
	t.Logf("poll 超時（%s），回最後一次狀態：status=%s", realChainPollTimeout, last.Status)
	return last
}

// containsModelID 檢查 model 清單是否含指定 id。
func containsModelID(models []*model.Model, id string) bool {
	for _, m := range models {
		if m.ID == id {
			return true
		}
	}
	return false
}

// ==========================================================================
// 給 Orchestrator 對 stage 跑（本機連不到 stage / 無 PG → 預設 Skip）
//
// 前置：
//   1. 取 converter API key（從 stage container env，不 hardcode）：
//        KEY=$(docker -H tcp://192.168.0.130:2375 exec \
//          kneron_model_converter-scheduler-1 printenv CONVERTER_API_KEY)
//   2. 取 PG DSN：指 stage 真 PG（已 migrate，含 users / models 表）。
//        若 stage PG 連線資訊未知，可在 stage 機器上跑、用 localhost DSN；
//        或本機起一個 docker PG 並對它跑 migrations/0001_create_users_models.up.sql。
//   3. fixture：onnx + 56 張 ref 圖（default 路徑見 real_converter_e2e_test.go；
//        env VISIONA_REAL_CONVERTER_ONNX / VISIONA_REAL_CONVERTER_IMAGES 可覆寫）。
//
// 跑：
//   VISIONA_REAL_CONVERTER_URL=http://192.168.0.130:9501 \
//   VISIONA_CONVERTER_API_KEY="$KEY" \
//   VISIONA_REAL_PG_DSN="postgres://visiona:<pw>@<pg-host>:5432/visiona?sslmode=disable" \
//   go test -tags=realconv ./internal/conversion/ \
//     -run TestRealChain_ConvertPromoteToPGModelLibrary -count=1 -v -timeout=20m
//
// 注意：真 KTC 轉檔較慢，-timeout 給 20m（poll deadline 內建 10m）。
// 測完會軟刪除建立的 model（t.Cleanup）；test user 留在 PG（無害、固定 UUID 可重用）。
// ==========================================================================