package control

import (
	"cloud.google.com/go/storage"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"google.golang.org/api/googleapi"
	"io"
	"net/url"
	"strconv"
	"strings"
	"sync"
	"time"
)

// This is current artifact evidence, not an immutable loader-generation receipt
// and never a target-table or downstream business acceptance result.
type ManifestEvidence struct {
	URI         string    `json:"uri,omitempty"`
	Generation  string    `json:"generation,omitempty"`
	SHA256      string    `json:"sha256,omitempty"`
	RunID       string    `json:"run_id"`
	SourceTable string    `json:"source_table"`
	Status      string    `json:"status"`
	Message     string    `json:"message"`
	ObservedAt  time.Time `json:"observed_at"`
}

type manifestArtifact struct {
	Body       []byte
	Generation int64
	Updated    time.Time
	Failure    string
}

func manifestReadFailure(stage string, err error) manifestArtifact {
	kind := "unavailable"
	var apiErr *googleapi.Error
	if errors.Is(err, context.DeadlineExceeded) {
		kind = "deadline"
	} else if errors.Is(err, storage.ErrObjectNotExist) {
		kind = "not_found"
	} else if errors.As(err, &apiErr) && (apiErr.Code == 401 || apiErr.Code == 403) {
		kind = "permission"
	}
	return manifestArtifact{Failure: stage + "_" + kind}
}

func fetchManifestArtifacts(ctx context.Context, requests map[string]string, read func(context.Context, string, string) (manifestArtifact, error)) (map[string]manifestArtifact, error) {
	artifacts := map[string]manifestArtifact{}
	var mu sync.Mutex
	var workers sync.WaitGroup
	slots := make(chan struct{}, 4)
	for uri, runID := range requests {
		uri, runID := uri, runID
		workers.Add(1)
		go func() {
			defer workers.Done()
			select {
			case slots <- struct{}{}:
			case <-ctx.Done():
				return
			}
			defer func() { <-slots }()
			call, cancel := context.WithTimeout(ctx, 15*time.Second)
			defer cancel()
			artifact, _ := read(call, uri, runID)
			mu.Lock()
			artifacts[uri] = artifact
			mu.Unlock()
		}()
	}
	workers.Wait()
	if ctx.Err() != nil {
		return nil, fmt.Errorf("CRM manifest collection deadline exceeded; checkpoint unchanged")
	}
	return artifacts, nil
}

func validatedCRMManifestURI(raw, runID string) (*url.URL, error) {
	u, err := url.Parse(raw)
	if err != nil || u.Scheme != "gs" || u.Host != "gcp-digital-bi-pingcap-nexuscrm-ingestion" || u.User != nil || u.RawQuery != "" || u.Fragment != "" || !strings.HasPrefix(u.Path, "/manifests/source=nexuscrm/") || runID == "" || strings.ContainsAny(runID, "/\\") || !strings.HasSuffix(u.Path, "/run_id="+runID+"/manifest.json") {
		return nil, fmt.Errorf("CRM manifest reference is outside the approved namespace")
	}
	return u, nil
}

func (s *ExternalLiveSource) readCRMManifest(ctx context.Context, uri, runID string) (manifestArtifact, error) {
	u, err := validatedCRMManifestURI(uri, runID)
	if err != nil || s.gcs == nil {
		return manifestArtifact{}, fmt.Errorf("CRM manifest reader unavailable or invalid scope")
	}
	object := s.gcs.Bucket(u.Host).Object(strings.TrimPrefix(u.Path, "/"))
	attrs, err := object.Attrs(ctx)
	const maxBytes = 8 << 20
	if err != nil {
		return manifestReadFailure("metadata", err), fmt.Errorf("CRM manifest metadata unavailable")
	}
	if attrs.Size < 1 || attrs.Size > maxBytes {
		return manifestArtifact{Failure: "size_limit"}, fmt.Errorf("CRM manifest metadata exceeds limit")
	}
	reader, err := object.Generation(attrs.Generation).NewReader(ctx)
	if err != nil {
		return manifestReadFailure("generation", err), fmt.Errorf("CRM manifest generation unavailable")
	}
	defer reader.Close()
	body, err := io.ReadAll(io.LimitReader(reader, maxBytes+1))
	if err != nil {
		return manifestReadFailure("content", err), fmt.Errorf("CRM manifest read incomplete")
	}
	if int64(len(body)) != attrs.Size {
		return manifestArtifact{Failure: "size_mismatch"}, fmt.Errorf("CRM manifest read incomplete")
	}
	return manifestArtifact{Body: body, Generation: attrs.Generation, Updated: attrs.Updated}, nil
}

func evaluateManifest(uri, runID, sourceTable string, loadStarted, auditedAt, now time.Time, artifact manifestArtifact) ManifestEvidence {
	e := ManifestEvidence{RunID: runID, SourceTable: sourceTable, Status: "unavailable", Message: "manifest 未读取；加载审计不等于提取完成或目标验收", ObservedAt: now}
	if _, err := validatedCRMManifestURI(uri, runID); err != nil {
		e.Status = "unbound"
		e.Message = "owner manifest 指针缺失、歧义或超出批准来源范围"
		return e
	}
	e.URI = uri
	if len(artifact.Body) == 0 || artifact.Generation <= 0 || artifact.Updated.IsZero() {
		if artifact.Failure != "" {
			e.Message = "manifest 读取未完成（" + artifact.Failure + "）；保留加载审计，不推断提取或业务成败"
		}
		return e
	}
	e.Generation = strconv.FormatInt(artifact.Generation, 10)
	hash := sha256.Sum256(artifact.Body)
	e.SHA256 = hex.EncodeToString(hash[:])
	var manifest struct {
		Version int       `json:"manifest_schema_version"`
		Source  string    `json:"source_name"`
		Project string    `json:"target_project"`
		RunID   string    `json:"run_id"`
		URI     string    `json:"manifest_uri"`
		Status  string    `json:"product_status"`
		Created time.Time `json:"created_at"`
		Tables  []struct {
			Name string `json:"table_name"`
		} `json:"tables"`
	}
	if json.Unmarshal(artifact.Body, &manifest) != nil || (manifest.Version != 2 && manifest.Version != 3) || manifest.Source != "nexuscrm" || manifest.Project != "gcp-digital-bi" || manifest.RunID != runID || manifest.URI != uri || manifest.Status != "exported" || manifest.Created.IsZero() {
		e.Status = "mismatch"
		e.Message = "manifest 版本、来源、身份或导出声明不匹配；不可作为上游输入证据"
		return e
	}
	matches := 0
	for _, table := range manifest.Tables {
		if table.Name == sourceTable {
			matches++
		}
	}
	if matches != 1 {
		e.Status = "mismatch"
		e.Message = "manifest 未唯一声明该源表；不可绑定本次逐表加载"
		return e
	}
	if loadStarted.IsZero() || auditedAt.IsZero() || loadStarted.After(auditedAt) || artifact.Updated.After(loadStarted) || manifest.Created.After(loadStarted) {
		e.Status = "current_pointer_only"
		e.Message = "当前 manifest 可读，但不能证明此版本在该次加载前已存在"
		return e
	}
	e.Status = "consistent"
	e.Message = "owner run、表声明及当前版本时间一致；未取得加载时 generation 回执，未验证目标内容或下游业务"
	return e
}