package control

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"net/url"
	"os"
	"path"
	"regexp"
	"strings"
	"sync"
	"time"

	"cloud.google.com/go/bigquery"
	"cloud.google.com/go/storage"
	mysqlDriver "github.com/go-sql-driver/mysql"
	"google.golang.org/api/cloudscheduler/v1"
	"google.golang.org/api/iterator"
	loggingapi "google.golang.org/api/logging/v2"
	runapi "google.golang.org/api/run/v2"
)

var qualifiedTablePattern = regexp.MustCompile(`^[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+$`)

type LiveSourceConfig struct {
	FivetranAPIKey       string
	FivetranAPISecret    string
	FivetranConnectorID  string
	QueryProject         string
	BillingInputTable    string
	MonthlyOutputDataset string
	CRMControlDataset    string
	FinanceDatabaseURL   string
	FinanceServiceURL    string
	FinanceHistoryAPIURL string
	SchedulerJobName     string
}

func LiveSourceConfigFromEnv() LiveSourceConfig {
	return LiveSourceConfig{
		FivetranAPIKey:       os.Getenv("FIVETRAN_API_KEY"),
		FivetranAPISecret:    os.Getenv("FIVETRAN_API_SECRET"),
		FivetranConnectorID:  envOrValue("FIVETRAN_CONNECTOR_ID", "spake_desecrate"),
		QueryProject:         envOrValue("FINANCE_PORTAL_OBSERVATION_PROJECT", "gcp-digital-bi"),
		BillingInputTable:    envOrValue("FINANCE_PORTAL_AR_TABLE", "gcp-digital-bi.google_sheets.org_monthly_bill_confirmation_for_revrec"),
		MonthlyOutputDataset: envOrValue("FINANCE_PORTAL_MONTHLY_OUTPUT_DATASET", "gcp-digital-bi-dev.sandbox_dwh_ywang"),
		CRMControlDataset:    envOrValue("FINANCE_PORTAL_CRM_CONTROL_DATASET", "gcp-digital-bi.nexuscrm_control"),
		FinanceDatabaseURL:   os.Getenv("FINANCE_PORTAL_FINANCE_SOURCE_DSN"),
		FinanceServiceURL:    envOrValue("FINANCE_PORTAL_FINANCE_SERVICE_URL", "https://data-platform-finance-rqyfcmbm3q-de.a.run.app"),
		FinanceHistoryAPIURL: os.Getenv("FINANCE_PORTAL_FINANCE_HISTORY_API_URL"),
		SchedulerJobName:     envOrValue("FINANCE_PORTAL_FINANCE_SCHEDULER", "projects/gcp-digital-bi/locations/asia-east1/jobs/data-platform-finance-daily"),
	}
}

type ExternalLiveSource struct {
	config       LiveSourceConfig
	bq           *bigquery.Client
	finance      *sql.DB
	scheduler    *cloudscheduler.Service
	logging      *loggingapi.Service
	runJobs      *runapi.Service
	gcs          *storage.Client
	httpClient   *http.Client
	startupError map[string]bool
	logger       *slog.Logger
}

func NewExternalLiveSource(ctx context.Context, config LiveSourceConfig, logger *slog.Logger) *ExternalLiveSource {
	if logger == nil {
		logger = slog.Default()
	}
	source := &ExternalLiveSource{
		config:       config,
		httpClient:   &http.Client{Timeout: 15 * time.Second, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }},
		startupError: make(map[string]bool),
		logger:       logger,
	}
	if client, err := bigquery.NewClient(ctx, config.QueryProject); err == nil {
		source.bq = client
	} else {
		source.startupError["bigquery"] = true
	}
	if config.FinanceDatabaseURL != "" {
		if dsn, err := mysqlURLToDSN(config.FinanceDatabaseURL); err == nil {
			if database, openErr := sql.Open("mysql", dsn); openErr == nil {
				database.SetMaxOpenConns(4)
				database.SetMaxIdleConns(2)
				database.SetConnMaxLifetime(5 * time.Minute)
				source.finance = database
			} else {
				source.startupError["finance_tidb"] = true
			}
		} else {
			source.startupError["finance_tidb"] = true
		}
	}
	if service, err := cloudscheduler.NewService(ctx); err == nil {
		source.scheduler = service
	} else {
		source.startupError["finance_scheduler"] = true
	}
	if service, err := loggingapi.NewService(ctx); err == nil {
		source.logging = service
	} else {
		source.startupError["finance_scheduler_logs"] = true
	}
	if client, err := storage.NewClient(ctx); err == nil {
		source.gcs = client
	}
	if client, err := runapi.NewService(ctx); err == nil {
		source.runJobs = client
	}
	return source
}

func (s *ExternalLiveSource) Close() error {
	var first error
	if s.gcs != nil {
		first = s.gcs.Close()
	}
	if s.bq != nil {
		if err := s.bq.Close(); err != nil {
			first = err
		}
	}
	if s.finance != nil {
		if err := s.finance.Close(); err != nil && first == nil {
			first = err
		}
	}
	return first
}

func (s *ExternalLiveSource) Read(ctx context.Context, billMonth, mode string) LiveFacts {
	facts := LiveFacts{ObservedAt: time.Now().UTC()}
	var mu sync.Mutex
	var wait sync.WaitGroup
	probe := func(read func() (string, DependencyObservation, any), assign func(any)) {
		defer wait.Done()
		key, dependency, result := read()
		mu.Lock()
		facts.Dependencies = append(facts.Dependencies, dependency)
		assign(result)
		mu.Unlock()
		if dependency.Status == "unavailable" {
			s.logger.Warn("live-dependency-unavailable", "dependency", key)
		}
	}

	wait.Add(7)
	go probe(func() (string, DependencyObservation, any) {
		fact, dep := s.readFivetran(ctx)
		return dep.Key, dep, fact
	}, func(value any) { facts.Fivetran = value.(FivetranFact) })
	go probe(func() (string, DependencyObservation, any) {
		fact, dep := s.readBillingInput(ctx, billMonth)
		return dep.Key, dep, fact
	}, func(value any) { facts.BillingInput = value.(BillingInputFact) })
	go probe(func() (string, DependencyObservation, any) {
		fact, dep := s.readMonthlyOutput(ctx, billMonth, mode)
		return dep.Key, dep, fact
	}, func(value any) { facts.MonthlyOutput = value.(MonthlyOutputFact) })
	go probe(func() (string, DependencyObservation, any) {
		fact, dep := s.readFinanceClose(ctx, billMonth)
		return dep.Key, dep, fact
	}, func(value any) { facts.Finance = value.(FinanceCloseFact) })
	go probe(func() (string, DependencyObservation, any) { fact, dep := s.readCRM(ctx); return dep.Key, dep, fact }, func(value any) { facts.CRM = value.(CRMPipelineFact) })
	go probe(func() (string, DependencyObservation, any) {
		fact, dep := s.readSalesforceOrder(ctx)
		return dep.Key, dep, fact
	}, func(value any) { facts.Salesforce = value.(SalesforceOrderFact) })
	go probe(func() (string, DependencyObservation, any) {
		schedulerFact, schedulerDependency := s.readScheduler(ctx)
		serviceFact, serviceDependency := s.readFinanceService(ctx)
		mu.Lock()
		facts.Dependencies = append(facts.Dependencies, serviceDependency)
		facts.FinanceService = serviceFact
		mu.Unlock()
		return schedulerDependency.Key, schedulerDependency, schedulerFact
	}, func(value any) { facts.Scheduler = value.(SchedulerFact) })
	wait.Wait()
	return facts
}

func (s *ExternalLiveSource) readFivetran(ctx context.Context) (FivetranFact, DependencyObservation) {
	key := "fivetran"
	configured := s.config.FivetranAPIKey != "" && s.config.FivetranAPISecret != "" && s.config.FivetranConnectorID != ""
	if !configured {
		return FivetranFact{}, unavailableDependency(key, "Fivetran", "missing FIVETRAN_API_KEY or FIVETRAN_API_SECRET", false)
	}
	request, err := http.NewRequestWithContext(ctx, http.MethodGet, "https://api.fivetran.com/v1/connectors/"+url.PathEscape(s.config.FivetranConnectorID), nil)
	if err != nil {
		return FivetranFact{}, unavailableDependency(key, "Fivetran", "connector request could not be created", true)
	}
	request.SetBasicAuth(s.config.FivetranAPIKey, s.config.FivetranAPISecret)
	response, err := s.httpClient.Do(request)
	if err != nil {
		return FivetranFact{}, unavailableDependency(key, "Fivetran", "connector API is unavailable", true)
	}
	defer response.Body.Close()
	if response.StatusCode != http.StatusOK {
		_, _ = io.Copy(io.Discard, response.Body)
		return FivetranFact{}, unavailableDependency(key, "Fivetran", fmt.Sprintf("connector API returned HTTP %d", response.StatusCode), true)
	}
	var payload struct {
		Data struct {
			ID          string  `json:"id"`
			Paused      bool    `json:"paused"`
			SucceededAt *string `json:"succeeded_at"`
			FailedAt    *string `json:"failed_at"`
			Status      struct {
				SyncState   string `json:"sync_state"`
				UpdateState string `json:"update_state"`
			} `json:"status"`
		} `json:"data"`
	}
	if err := json.NewDecoder(io.LimitReader(response.Body, 1<<20)).Decode(&payload); err != nil {
		return FivetranFact{}, unavailableDependency(key, "Fivetran", "connector API response was invalid", true)
	}
	fact := FivetranFact{ConnectorID: payload.Data.ID, SyncState: payload.Data.Status.SyncState, UpdateState: payload.Data.Status.UpdateState, Paused: payload.Data.Paused, SucceededAt: parseOptionalTime(payload.Data.SucceededAt), FailedAt: parseOptionalTime(payload.Data.FailedAt)}
	status := "healthy"
	message := "connector scheduled and latest successful sync is available"
	if fact.Paused || fact.SucceededAt == nil {
		status = "degraded"
		message = "connector is paused or has no successful sync receipt"
	}
	return fact, observedDependency(key, "Fivetran", status, message, fact.SucceededAt, true)
}

func (s *ExternalLiveSource) readBillingInput(ctx context.Context, billMonth string) (BillingInputFact, DependencyObservation) {
	key := "bigquery_ar"
	if s.bq == nil || s.startupError["bigquery"] {
		return BillingInputFact{}, unavailableDependency(key, "BigQuery AR Table", "Application Default Credentials cannot access BigQuery", true)
	}
	if !qualifiedTablePattern.MatchString(s.config.BillingInputTable) {
		return BillingInputFact{}, unavailableDependency(key, "BigQuery AR Table", "FINANCE_PORTAL_AR_TABLE is invalid", true)
	}
	query := s.bq.Query(fmt.Sprintf(`
		SELECT
			COUNT(*) AS total_rows,
			COUNTIF(NULLIF(TRIM(actual_revenue_recognition_date), '') IS NOT NULL) AS actual_rows,
			COUNTIF(NULLIF(TRIM(actual_revenue_recognition_date), '') IS NULL) AS predicted_rows,
			MAX(_fivetran_synced) AS last_synced_at
		FROM %s
		WHERE bill_month = @bill_month`, quoteBQTable(s.config.BillingInputTable)))
	query.Parameters = []bigquery.QueryParameter{{Name: "bill_month", Value: billMonth}}
	var row struct {
		TotalRows     int64                  `bigquery:"total_rows"`
		ActualRows    int64                  `bigquery:"actual_rows"`
		PredictedRows int64                  `bigquery:"predicted_rows"`
		LastSyncedAt  bigquery.NullTimestamp `bigquery:"last_synced_at"`
	}
	if err := readBigQueryRow(ctx, query, &row); err != nil {
		return BillingInputFact{}, unavailableDependency(key, "BigQuery AR Table", "AR input query failed", true)
	}
	fact := BillingInputFact{TotalRows: row.TotalRows, ActualRows: row.ActualRows, PredictedRows: row.PredictedRows, LastSyncedAt: nullTimestamp(row.LastSyncedAt)}
	status := "healthy"
	message := fmt.Sprintf("%s has %d rows; actual=%d predicted=%d", billMonth, fact.TotalRows, fact.ActualRows, fact.PredictedRows)
	if fact.TotalRows == 0 {
		status = "degraded"
		message = billMonth + " has no landed AR rows"
	}
	return fact, observedDependency(key, "BigQuery AR Table", status, message, fact.LastSyncedAt, true)
}

func (s *ExternalLiveSource) readMonthlyOutput(ctx context.Context, billMonth, mode string) (MonthlyOutputFact, DependencyObservation) {
	key := "monthly_billing_output"
	if s.bq == nil || s.startupError["bigquery"] {
		return MonthlyOutputFact{}, unavailableDependency(key, "Monthly Billing Output", "Application Default Credentials cannot access BigQuery", true)
	}
	parts := strings.Split(s.config.MonthlyOutputDataset, ".")
	if len(parts) != 2 || !regexp.MustCompile(`^[A-Za-z0-9_-]+$`).MatchString(parts[0]) || !regexp.MustCompile(`^[A-Za-z0-9_-]+$`).MatchString(parts[1]) {
		return MonthlyOutputFact{}, unavailableDependency(key, "Monthly Billing Output", "FINANCE_PORTAL_MONTHLY_OUTPUT_DATASET is invalid", true)
	}
	suffix := strings.ReplaceAll(billMonth, "-", "_") + "_" + mode
	orders := s.tableMetadata(ctx, parts[0], parts[1], "monthly_revrec_orders_"+suffix)
	collections := s.tableMetadata(ctx, parts[0], parts[1], "monthly_revrec_collections_"+suffix)
	fact := MonthlyOutputFact{Orders: orders, Collections: collections}
	observedAt := latestTime(orders.LastModified, collections.LastModified)
	status := "healthy"
	message := fmt.Sprintf("orders=%d collections=%d", orders.Rows, collections.Rows)
	if orders.Rows == 0 || collections.Rows == 0 {
		status = "degraded"
		message = "one or both run outputs are missing"
	}
	return fact, observedDependency(key, "Monthly Billing Output", status, message, observedAt, true)
}

func (s *ExternalLiveSource) tableMetadata(ctx context.Context, projectID, datasetID, tableID string) TableFact {
	fact := TableFact{Name: strings.Join([]string{projectID, datasetID, tableID}, ".")}
	metadata, err := s.bq.DatasetInProject(projectID, datasetID).Table(tableID).Metadata(ctx)
	if err != nil {
		return fact
	}
	modified := metadata.LastModifiedTime.UTC()
	fact.Rows = int64(metadata.NumRows)
	fact.LastModified = &modified
	return fact
}

func (s *ExternalLiveSource) readFinanceClose(ctx context.Context, billMonth string) (FinanceCloseFact, DependencyObservation) {
	key := "finance_tidb"
	if s.config.FinanceDatabaseURL == "" {
		return FinanceCloseFact{}, unavailableDependency(key, "Finance TiDB", "missing FINANCE_PORTAL_FINANCE_SOURCE_DSN", false)
	}
	if s.finance == nil || s.startupError[key] {
		return FinanceCloseFact{}, unavailableDependency(key, "Finance TiDB", "read-only database connection is unavailable", true)
	}
	rows, err := s.finance.QueryContext(ctx, `
		SELECT 'order' AS kind, COUNT(*) AS total,
			COALESCE(SUM(sync = 1), 0), COALESCE(SUM(COALESCE(sync, 0) = 0), 0),
			COALESCE(SUM(NULLIF(TRIM(new_order_id), '') IS NOT NULL), 0),
			COALESCE(SUM(COALESCE(is_new_salesforce_order, 0) = 1), 0), 0, 0,
			COALESCE(SUM(COALESCE(row_lock, 0) = 1), 0),
			MAX(COALESCE(updated_at, created_at)), MAX(CASE WHEN sync = 1 THEN sync_time END)
		FROM monthly_cloud_order WHERE bill_month = ? AND deleted_at IS NULL
		UNION ALL
		SELECT 'revrec', COUNT(*),
			COALESCE(SUM(sync = 1), 0), COALESCE(SUM(COALESCE(sync, 0) = 0), 0),
			COALESCE(SUM(NULLIF(TRIM(revrec_id), '') IS NOT NULL), 0),
			0,
			COALESCE(SUM(LOWER(TRIM(COALESCE(revenue_recognition_type, ''))) = 'estimated' OR estimate = 1), 0),
			COALESCE(SUM(LOWER(TRIM(COALESCE(revenue_recognition_type, ''))) = 'actual'), 0),
			COALESCE(SUM(COALESCE(row_lock, 0) = 1), 0),
			MAX(COALESCE(updated_at, created_at)), MAX(CASE WHEN sync = 1 THEN sync_time END)
		FROM monthly_cloud_revrec WHERE bill_month = ? AND deleted_at IS NULL
		UNION ALL
		SELECT 'collection', COUNT(*),
			COALESCE(SUM(sync = 1), 0), COALESCE(SUM(COALESCE(sync, 0) = 0), 0),
			COALESCE(SUM(NULLIF(TRIM(collection_id), '') IS NOT NULL), 0), 0, 0, 0,
			COALESCE(SUM(COALESCE(row_lock, 0) = 1), 0),
			MAX(COALESCE(updated_at, created_at)), MAX(CASE WHEN sync = 1 THEN sync_time END)
		FROM monthly_cloud_collection WHERE bill_month = ? AND deleted_at IS NULL`, billMonth, billMonth, billMonth)
	if err != nil {
		return FinanceCloseFact{}, unavailableDependency(key, "Finance TiDB", "monthly close read failed", true)
	}
	defer rows.Close()
	fact := FinanceCloseFact{}
	for rows.Next() {
		var kind string
		var table FinanceTableFact
		var updatedAt, syncedAt sql.NullTime
		if err := rows.Scan(&kind, &table.Total, &table.Synced, &table.Unsynced, &table.IdentityRows, &table.RequiredIdentityRows, &table.EstimatedRows, &table.ActualRows, &table.LockedRows, &updatedAt, &syncedAt); err != nil {
			return FinanceCloseFact{}, unavailableDependency(key, "Finance TiDB", "monthly close result was invalid", true)
		}
		table.LastUpdated = nullSQLTime(updatedAt)
		table.LastSynced = nullSQLTime(syncedAt)
		switch kind {
		case "order":
			fact.Order = table
		case "revrec":
			fact.RevRec = table
		case "collection":
			fact.Collection = table
		}
	}
	if err := rows.Err(); err != nil {
		return FinanceCloseFact{}, unavailableDependency(key, "Finance TiDB", "monthly close rows could not be read", true)
	}
	var receiptAt sql.NullTime
	_ = s.finance.QueryRowContext(ctx, `SELECT id, created_at FROM error_logs WHERE severity = 1003 AND LOWER(error_message) LIKE '%cloud revrec%' ORDER BY id DESC LIMIT 1`).Scan(&fact.BreakdownReceiptID, &receiptAt)
	fact.BreakdownSucceededAt = nullSQLTime(receiptAt)
	observedAt := latestTime(fact.Order.LastUpdated, fact.RevRec.LastUpdated, fact.Collection.LastUpdated, fact.BreakdownSucceededAt)
	message := fmt.Sprintf("order=%d/%d revrec=%d/%d collection=%d/%d synced", fact.Order.Synced, fact.Order.Total, fact.RevRec.Synced, fact.RevRec.Total, fact.Collection.Synced, fact.Collection.Total)
	return fact, observedDependency(key, "Finance TiDB", "healthy", message, observedAt, true)
}

func (s *ExternalLiveSource) readCRM(ctx context.Context) (CRMPipelineFact, DependencyObservation) {
	key := "crm_pipeline"
	if s.bq == nil || s.startupError["bigquery"] {
		return CRMPipelineFact{}, unavailableDependency(key, "CRM Pipeline", "Application Default Credentials cannot access BigQuery", true)
	}
	dataset := s.config.CRMControlDataset
	if len(strings.Split(dataset, ".")) != 2 {
		return CRMPipelineFact{}, unavailableDependency(key, "CRM Pipeline", "FINANCE_PORTAL_CRM_CONTROL_DATASET is invalid", true)
	}
	query := s.bq.Query(fmt.Sprintf(`
		SELECT final_table, run_id, status, final_rows, duplicate_primary_key_rows, created_at
		FROM %s
		WHERE final_table IN ('nexuscrm_final.nexuscrm_order', 'nexuscrm_final.nexuscrm_orderitem', 'nexuscrm_final.nexuscrm_product2')
		QUALIFY ROW_NUMBER() OVER (PARTITION BY final_table ORDER BY created_at DESC) = 1`, quoteBQTable(dataset+".table_loads")))
	iteratorRows, err := query.Read(ctx)
	if err != nil {
		return CRMPipelineFact{}, unavailableDependency(key, "CRM Pipeline", "pipeline audit query failed", true)
	}
	fact := CRMPipelineFact{}
	for {
		var row struct {
			FinalTable string                 `bigquery:"final_table"`
			RunID      string                 `bigquery:"run_id"`
			Status     string                 `bigquery:"status"`
			FinalRows  int64                  `bigquery:"final_rows"`
			Duplicates int64                  `bigquery:"duplicate_primary_key_rows"`
			CreatedAt  bigquery.NullTimestamp `bigquery:"created_at"`
		}
		if err := iteratorRows.Next(&row); err == iterator.Done {
			break
		} else if err != nil {
			return CRMPipelineFact{}, unavailableDependency(key, "CRM Pipeline", "pipeline audit result failed", true)
		}
		table := CRMTableFact{RunID: row.RunID, Status: row.Status, Rows: row.FinalRows, DuplicatePKRows: row.Duplicates, ObservedAt: nullTimestamp(row.CreatedAt)}
		tableName := row.FinalTable[strings.LastIndex(row.FinalTable, ".")+1:]
		switch tableName {
		case "nexuscrm_order":
			fact.Order = table
		case "nexuscrm_orderitem":
			fact.OrderItem = table
		case "nexuscrm_product2":
			fact.Product2 = table
		}
	}
	observedAt := latestTime(fact.Order.ObservedAt, fact.OrderItem.ObservedAt, fact.Product2.ObservedAt)
	status := "healthy"
	message := fmt.Sprintf("Order=%d OrderItem=%d Product2=%d", fact.Order.Rows, fact.OrderItem.Rows, fact.Product2.Rows)
	if fact.Order.RunID == "" || fact.Order.RunID != fact.OrderItem.RunID || fact.Order.DuplicatePKRows > 0 || fact.OrderItem.DuplicatePKRows > 0 {
		status = "degraded"
		message = "latest Order and OrderItem receipts are missing, mismatched, or duplicated"
	}
	return fact, observedDependency(key, "CRM Pipeline", status, message, observedAt, true)
}

func (s *ExternalLiveSource) readSalesforceOrder(ctx context.Context) (SalesforceOrderFact, DependencyObservation) {
	key := "finance_salesforce_order"
	if s.config.FinanceDatabaseURL == "" {
		return SalesforceOrderFact{}, unavailableDependency(key, "Finance Salesforce Order", "missing FINANCE_PORTAL_FINANCE_SOURCE_DSN", false)
	}
	if s.finance == nil || s.startupError["finance_tidb"] {
		return SalesforceOrderFact{}, unavailableDependency(key, "Finance Salesforce Order", "read-only audit connection is unavailable", true)
	}
	var fact SalesforceOrderFact
	var finishedAt sql.NullTime
	err := s.finance.QueryRowContext(ctx, `
		SELECT run_id, status, COALESCE(source_row_count, 0), COALESCE(target_after_count, 0),
			COALESCE(duplicate_key_count, 0), COALESCE(critical_null_count, 0), finished_at
		FROM data_platform_load_table_audit
		WHERE target_table = 'salesforce_order' AND status IN ('success', 'warning')
		ORDER BY finished_at DESC LIMIT 1`).Scan(&fact.RunID, &fact.Status, &fact.SourceRows, &fact.TargetRows, &fact.DuplicateKeyRows, &fact.CriticalNullRows, &finishedAt)
	if err != nil {
		return SalesforceOrderFact{}, unavailableDependency(key, "Finance Salesforce Order", "latest table audit could not be read", true)
	}
	fact.FinishedAt = nullSQLTime(finishedAt)
	status := "healthy"
	message := fmt.Sprintf("latest receipt %s loaded %d rows", fact.RunID, fact.TargetRows)
	if fact.DuplicateKeyRows > 0 || fact.CriticalNullRows > 0 {
		status = "degraded"
		message = "latest receipt has duplicate or critical-null findings"
	}
	return fact, observedDependency(key, "Finance Salesforce Order", status, message, fact.FinishedAt, true)
}

func (s *ExternalLiveSource) readScheduler(ctx context.Context) (SchedulerFact, DependencyObservation) {
	key := "finance_scheduler"
	if s.scheduler == nil || s.startupError[key] {
		return SchedulerFact{}, unavailableDependency(key, "Finance Scheduler", "Cloud Scheduler API is unavailable", true)
	}
	job, err := s.scheduler.Projects.Locations.Jobs.Get(s.config.SchedulerJobName).Context(ctx).Do()
	if err != nil {
		return SchedulerFact{}, unavailableDependency(key, "Finance Scheduler", "scheduler job could not be read", true)
	}
	fact := SchedulerFact{Name: path.Base(job.Name), State: job.State, Schedule: job.Schedule, TimeZone: job.TimeZone, NextRunAt: parseTimeString(job.ScheduleTime), LastAttemptAt: parseTimeString(job.LastAttemptTime)}
	if job.Status != nil {
		fact.LastStatusCode = job.Status.Code
	}
	if job.HttpTarget != nil {
		fact.Target = job.HttpTarget.Uri
		fact.HasExplicitBody = strings.TrimSpace(job.HttpTarget.Body) != ""
	}
	return fact, schedulerDependency(fact, time.Now().UTC())
}

func schedulerDependency(fact SchedulerFact, now time.Time) DependencyObservation {
	status := "healthy"
	message := fmt.Sprintf("%s; last status code=%d; next run %s", fact.State, fact.LastStatusCode, displayTime(fact.NextRunAt))
	if fact.State != "ENABLED" || fact.LastStatusCode != 0 || fact.LastAttemptAt == nil {
		status = "degraded"
	}
	if fact.State == "ENABLED" && !fact.HasExplicitBody {
		status = "degraded"
		message = fmt.Sprintf("ENABLED with empty body; last status code=%d; next run %s; legacy /run defaults to all Finance tasks", fact.LastStatusCode, displayTime(fact.NextRunAt))
	}
	// Poll observation time is not the future planned execution time.
	return observedDependency("finance_scheduler", "Finance Scheduler", status, message, &now, true)
}

func (s *ExternalLiveSource) readFinanceService(ctx context.Context) (ServiceFact, DependencyObservation) {
	key := "finance_service"
	if s.config.FinanceServiceURL == "" {
		return ServiceFact{}, unavailableDependency(key, "data-platform-finance", "FINANCE_PORTAL_FINANCE_SERVICE_URL is missing", false)
	}
	request, err := http.NewRequestWithContext(ctx, http.MethodGet, strings.TrimRight(s.config.FinanceServiceURL, "/")+"/health", nil)
	if err != nil {
		return ServiceFact{}, unavailableDependency(key, "data-platform-finance", "health request could not be created", true)
	}
	response, err := s.httpClient.Do(request)
	if err != nil {
		return ServiceFact{}, unavailableDependency(key, "data-platform-finance", "health endpoint is unavailable", true)
	}
	defer response.Body.Close()
	_, _ = io.Copy(io.Discard, io.LimitReader(response.Body, 1<<16))
	now := time.Now().UTC()
	fact := ServiceFact{URL: s.config.FinanceServiceURL, Healthy: response.StatusCode == http.StatusOK, ObservedAt: &now}
	status := "healthy"
	message := "Cloud Run health returned HTTP 200"
	if !fact.Healthy {
		status = "degraded"
		message = fmt.Sprintf("Cloud Run health returned HTTP %d", response.StatusCode)
	}
	return fact, observedDependency(key, "data-platform-finance", status, message, &now, true)
}

func readBigQueryRow(ctx context.Context, query *bigquery.Query, destination any) error {
	rows, err := query.Read(ctx)
	if err != nil {
		return err
	}
	return rows.Next(destination)
}

func mysqlURLToDSN(raw string) (string, error) {
	parsed, err := url.Parse(raw)
	if err != nil || parsed.Scheme != "mysql" || parsed.User == nil || parsed.Host == "" {
		return "", fmt.Errorf("invalid mysql URL")
	}
	password, _ := parsed.User.Password()
	database, err := url.PathUnescape(strings.TrimPrefix(parsed.Path, "/"))
	if err != nil || database == "" {
		return "", fmt.Errorf("mysql database is missing")
	}
	config := mysqlDriver.NewConfig()
	config.User = parsed.User.Username()
	config.Passwd = password
	config.Net = "tcp"
	config.Addr = parsed.Host
	config.DBName = database
	config.ParseTime = true
	config.Loc = time.UTC
	if ssl := strings.ToLower(parsed.Query().Get("ssl")); ssl == "true" || ssl == "1" || ssl == "required" {
		config.TLSConfig = "true"
	}
	return config.FormatDSN(), nil
}

func quoteBQTable(value string) string { return "`" + value + "`" }

func nullTimestamp(value bigquery.NullTimestamp) *time.Time {
	if !value.Valid {
		return nil
	}
	result := value.Timestamp.UTC()
	return &result
}

func nullSQLTime(value sql.NullTime) *time.Time {
	if !value.Valid {
		return nil
	}
	result := value.Time.UTC()
	return &result
}

func parseOptionalTime(value *string) *time.Time {
	if value == nil {
		return nil
	}
	return parseTimeString(*value)
}

func parseTimeString(value string) *time.Time {
	if value == "" {
		return nil
	}
	parsed, err := time.Parse(time.RFC3339Nano, value)
	if err != nil {
		return nil
	}
	parsed = parsed.UTC()
	return &parsed
}

func observedDependency(key, label, status, message string, observedAt *time.Time, configured bool) DependencyObservation {
	return DependencyObservation{Key: key, Label: label, Status: status, Message: message, ObservedAt: observedAt, CredentialConfigured: configured}
}

func unavailableDependency(key, label, message string, configured bool) DependencyObservation {
	return observedDependency(key, label, "unavailable", message, nil, configured)
}

func displayTime(value *time.Time) string {
	if value == nil {
		return "unknown"
	}
	return value.Format(time.RFC3339)
}

func envOrValue(key, fallback string) string {
	if value := strings.TrimSpace(os.Getenv(key)); value != "" {
		return value
	}
	return fallback
}