package lake

import (
	"bytes"
	"context"
	"crypto/hmac"
	"crypto/rand"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"sort"
	"strconv"
	"strings"
	"time"
)

const (
	tokenExchangePath  = "/api/v1/internal/tidb-cloud/token-exchange"
	businessPathPrefix = "/api/v1/tidb-cloud/v2/data-integration"

	maxTokenResponseBytes    = 64 << 10
	maxBusinessRequestBytes  = 1 << 20
	maxBusinessResponseBytes = 8 << 20
	listPageSize             = 100
)

type IDGenerator func() (string, error)

type HTTPClientConfig struct {
	BaseURL    string
	HMACSecret Secret
	HTTPClient *http.Client
	Clock      func() time.Time
	Nonce      IDGenerator
	RequestID  IDGenerator
}

type HTTPClient struct {
	baseURL    string
	hmacSecret Secret
	httpClient *http.Client
	clock      func() time.Time
	nonce      IDGenerator
	requestID  IDGenerator
}

var _ Client = (*HTTPClient)(nil)

func NewHTTPClient(config HTTPClientConfig) (*HTTPClient, error) {
	baseURL, err := validateBaseURL(config.BaseURL)
	if err != nil {
		return nil, err
	}
	if config.HMACSecret.Empty() {
		return nil, errors.New("Lake HMAC secret is required")
	}
	httpClient := config.HTTPClient
	if httpClient == nil {
		httpClient = &http.Client{Timeout: 30 * time.Second}
	}
	httpClientCopy := *httpClient
	// Lake endpoints are fixed. Following redirects could replay a write or
	// forward the short-lived bearer token to an unexpected origin.
	httpClientCopy.CheckRedirect = func(_ *http.Request, _ []*http.Request) error {
		return http.ErrUseLastResponse
	}
	clock := config.Clock
	if clock == nil {
		clock = time.Now
	}
	nonce := config.Nonce
	if nonce == nil {
		nonce = randomUUID
	}
	requestID := config.RequestID
	if requestID == nil {
		requestID = randomUUID
	}
	return &HTTPClient{
		baseURL:    baseURL,
		hmacSecret: config.HMACSecret,
		httpClient: &httpClientCopy,
		clock:      clock,
		nonce:      nonce,
		requestID:  requestID,
	}, nil
}

func validateBaseURL(value string) (string, error) {
	parsed, err := url.Parse(value)
	if err != nil {
		return "", fmt.Errorf("parse Lake base URL: %w", err)
	}
	if (parsed.Scheme != "http" && parsed.Scheme != "https") || parsed.Host == "" {
		return "", errors.New("Lake base URL must be an HTTP(S) origin")
	}
	if parsed.User != nil || parsed.RawQuery != "" || parsed.Fragment != "" || (parsed.Path != "" && parsed.Path != "/") {
		return "", errors.New("Lake base URL must not contain credentials, path, query, or fragment")
	}
	// Return the reconstructed origin rather than the caller's string: a bare
	// "?" or "#" suffix parses clean but would otherwise be concatenated in
	// front of the path, sending the request somewhere the signature does not
	// cover.
	return parsed.Scheme + "://" + parsed.Host, nil
}

// SetupInfo returns the identities a customer-owned IAM role must trust, and
// the external ID to bind them to, before role-ARN staging access works.
func (c *HTTPClient) SetupInfo(ctx context.Context, invocation Invocation) (SetupInfo, error) {
	request := wireSetupInfoRequest{}
	var response wireSetupInfoResponse
	if err := c.callBusiness(
		ctx, invocation, "/tidb/setup-info", &request, &response, false, "get setup info", nil,
	); err != nil {
		return SetupInfo{}, err
	}
	info := SetupInfo{
		ExternalID:      response.Data.ExternalID,
		TrustPrincipals: make([]TrustPrincipal, 0, len(response.Data.TrustPrincipals)),
	}
	for _, principal := range response.Data.TrustPrincipals {
		info.TrustPrincipals = append(info.TrustPrincipals, TrustPrincipal{
			Name: principal.Name, ARN: principal.ARN, Required: principal.Required,
		})
	}
	return info, nil
}

func (c *HTTPClient) TestDataSource(ctx context.Context, invocation Invocation, spec DataSourceSpec) error {
	request, secrets, err := c.buildDataSourceRequest(0, spec)
	if err != nil {
		return err
	}
	return c.callBusiness(ctx, invocation, "/datasource/test", &request, &wireSuccessResponse{}, false, "test datasource", secrets)
}

func (c *HTTPClient) CreateDataSource(ctx context.Context, invocation Invocation, spec DataSourceSpec) error {
	request, secrets, err := c.buildDataSourceRequest(0, spec)
	if err != nil {
		return err
	}
	return c.callBusiness(ctx, invocation, "/datasource/create", &request, &wireCreateResponse{}, true, "create datasource", secrets)
}

// GetDataSource looks a DataSource up by its unique name. A missing resource is
// a NotFound APIError, so callers check existence the same way they do for
// every other child system.
//
// It resolves the name against the list and then reads the record itself,
// rather than returning the list row. The list projection is thinner than
// /info - it omits the S3 endpoint and role ARN - and callers compare exactly
// those fields to decide whether a resource matches their spec, so a thin
// record would read as a permanent mismatch.
func (c *HTTPClient) GetDataSource(ctx context.Context, invocation Invocation, name string) (DataSource, error) {
	dataSourceID, err := c.dataSourceID(ctx, invocation, name)
	if err != nil {
		return DataSource{}, err
	}
	request := wireGetDataSourceRequest{DataSourceID: dataSourceID}
	var response wireDataSourceInfoResponse
	path := fmt.Sprintf("/datasource/%d/info", dataSourceID)
	if err := c.callBusiness(ctx, invocation, path, &request, &response, false, "get datasource", nil); err != nil {
		return DataSource{}, err
	}
	dataSource, err := dataSourceFromWire(response.Data)
	if err != nil {
		return DataSource{}, err
	}
	if dataSource.ID == 0 {
		dataSource.ID = dataSourceID
	}
	return dataSource, nil
}

// dataSourceID resolves the wire identifier Lake assigned to a name. Callers
// never see it: it exists only because Lake's own paths are numeric.
func (c *HTTPClient) dataSourceID(ctx context.Context, invocation Invocation, name string) (int32, error) {
	if strings.TrimSpace(name) == "" {
		return 0, errors.New("datasource name is required")
	}
	for page := int32(1); ; page++ {
		request := wireListDataSourcesRequest{
			Page: page, PerPage: listPageSize, ServiceID: ServiceIDTiDBCDC,
		}
		var response wireListDataSourcesResponse
		if err := c.callBusiness(ctx, invocation, "/datasources", &request, &response, false, "list datasources", nil); err != nil {
			return 0, err
		}
		for _, item := range response.Data.List {
			if item.Name == name {
				if item.ID != 0 {
					return item.ID, nil
				}
				return item.DataSourceID, nil
			}
		}
		// A short page is the last page. Trusting Total would stop early if
		// Lake reported it per-page, and a false NotFound makes the caller
		// create a duplicate of the resource it was looking for.
		if len(response.Data.List) < listPageSize {
			break
		}
	}
	return 0, notFoundError("datasource " + name)
}

func (c *HTTPClient) UpdateDataSource(ctx context.Context, invocation Invocation, name string, spec DataSourceSpec) error {
	dataSourceID, err := c.dataSourceID(ctx, invocation, name)
	if err != nil {
		return err
	}
	request, secrets, err := c.buildDataSourceRequest(dataSourceID, spec)
	if err != nil {
		return err
	}
	path := fmt.Sprintf("/datasource/%d/update", dataSourceID)
	return c.callBusiness(ctx, invocation, path, &request, &wireSuccessResponse{}, true, "update datasource", secrets)
}

func (c *HTTPClient) DeleteDataSource(ctx context.Context, invocation Invocation, name string) error {
	dataSourceID, err := c.dataSourceID(ctx, invocation, name)
	if err != nil {
		// A datasource that is not there is the state Delete converges to, so a
		// retry after a delete whose result was lost must not fail here.
		if isNotFound(err) {
			return nil
		}
		return err
	}
	request := wireDeleteDataSourceRequest{DataSourceIDs: []int32{dataSourceID}}
	return c.callBusiness(ctx, invocation, "/datasource/delete", &request, &wireSuccessResponse{}, true, "delete datasource", nil)
}

func (c *HTTPClient) CreateTask(ctx context.Context, invocation Invocation, spec TaskSpec) error {
	// Validate before resolving the datasource name: resolution is an HTTP
	// call, and a malformed spec must fail without touching the network.
	if err := validateTaskSpecForWire(spec); err != nil {
		return err
	}
	dataSourceID, err := c.dataSourceID(ctx, invocation, spec.DataSourceName)
	if err != nil {
		return err
	}
	request, err := c.buildTaskRequest(0, dataSourceID, spec)
	if err != nil {
		return err
	}
	return c.callBusiness(ctx, invocation, "/task/create", &request, &wireCreateResponse{}, true, "create task", nil)
}

// GetTask is the Task counterpart of GetDataSource, and reads the record itself
// for the same reason.
func (c *HTTPClient) GetTask(ctx context.Context, invocation Invocation, name string) (Task, error) {
	taskID, err := c.taskID(ctx, invocation, name)
	if err != nil {
		return Task{}, err
	}
	request := wireGetTaskRequest{TaskID: taskID}
	var response wireTaskInfoResponse
	path := fmt.Sprintf("/task/%d/info", taskID)
	if err := c.callBusiness(ctx, invocation, path, &request, &response, false, "get task", nil); err != nil {
		return Task{}, err
	}
	task := taskFromWire(response.Data)
	if task.ID == 0 {
		task.ID = taskID
	}
	return task, nil
}

func (c *HTTPClient) taskID(ctx context.Context, invocation Invocation, name string) (int32, error) {
	if strings.TrimSpace(name) == "" {
		return 0, errors.New("task name is required")
	}
	for page := int32(1); ; page++ {
		request := wireListTasksRequest{
			Page: page, PerPage: listPageSize, ServiceID: ServiceIDTiDBCDC,
		}
		var response wireListTasksResponse
		if err := c.callBusiness(ctx, invocation, "/tasks", &request, &response, false, "list tasks", nil); err != nil {
			return 0, err
		}
		for _, item := range response.Data.List {
			if item.Name == name {
				if item.ID != 0 {
					return item.ID, nil
				}
				return item.TaskID, nil
			}
		}
		if len(response.Data.List) < listPageSize {
			break
		}
	}
	return 0, notFoundError("task " + name)
}

func (c *HTTPClient) UpdateTask(ctx context.Context, invocation Invocation, name string, spec TaskSpec) error {
	if err := validateTaskSpecForWire(spec); err != nil {
		return err
	}
	task, err := c.GetTask(ctx, invocation, name)
	if err != nil {
		return err
	}
	// Resolve the datasource the spec asks for rather than reusing the one the
	// task already points at: TaskSpec says the adapter resolves the name, so
	// silently keeping the old datasource would let an update report success
	// while the task kept ingesting from the previous bucket and database.
	dataSourceID, err := c.dataSourceID(ctx, invocation, spec.DataSourceName)
	if err != nil {
		return err
	}
	request, err := c.buildTaskRequest(task.ID, dataSourceID, spec)
	if err != nil {
		return err
	}
	path := fmt.Sprintf("/task/%d/update", task.ID)
	return c.callBusiness(ctx, invocation, path, &request, &wireSuccessResponse{}, true, "update task", nil)
}

func (c *HTTPClient) StartTask(ctx context.Context, invocation Invocation, name string) error {
	return c.operateTask(ctx, invocation, name, "Start")
}

func (c *HTTPClient) StopTask(ctx context.Context, invocation Invocation, name string) error {
	return c.operateTask(ctx, invocation, name, "Stop")
}

func (c *HTTPClient) DropTask(ctx context.Context, invocation Invocation, name string) error {
	return c.operateTask(ctx, invocation, name, "Drop")
}

func (c *HTTPClient) operateTask(ctx context.Context, invocation Invocation, name, operation string) error {
	task, err := c.GetTask(ctx, invocation, name)
	if err != nil {
		return err
	}
	request := wireOperateTaskRequest{TaskIDs: []int32{task.ID}, Operate: operation}
	return c.callBusiness(ctx, invocation, "/task/operate", &request, &wireSuccessResponse{}, true, strings.ToLower(operation)+" task", nil)
}

func (c *HTTPClient) ListTaskRuns(ctx context.Context, invocation Invocation, name string, options ListTaskRunsOptions) (TaskRunPage, error) {
	task, err := c.GetTask(ctx, invocation, name)
	if err != nil {
		return TaskRunPage{}, err
	}
	return c.listTaskRuns(ctx, invocation, task.ID, options)
}

func (c *HTTPClient) listTaskRuns(ctx context.Context, invocation Invocation, taskID int32, options ListTaskRunsOptions) (TaskRunPage, error) {
	if options.Page <= 0 {
		options.Page = 1
	}
	if options.PerPage <= 0 {
		options.PerPage = 20
	}
	request := wireListTaskRunsRequest{
		TaskID: taskID, Page: options.Page, PerPage: options.PerPage,
	}
	var response wireListTaskRunsResponse
	if err := c.callBusiness(ctx, invocation, "/task/runs", &request, &response, false, "list task runs", nil); err != nil {
		return TaskRunPage{}, err
	}
	runs := make([]TaskRun, 0, len(response.Data.List))
	for _, item := range response.Data.List {
		run := taskRunFromWire(item)
		if run.TaskID == 0 {
			run.TaskID = taskID
		}
		runs = append(runs, run)
	}
	return TaskRunPage{Runs: runs, Total: response.Data.Total}, nil
}

func (c *HTTPClient) GetTaskRun(ctx context.Context, invocation Invocation, executeID string) (TaskRun, error) {
	if strings.TrimSpace(executeID) == "" {
		return TaskRun{}, errors.New("execute id is required")
	}
	request := wireGetTaskRunRequest{ExecuteID: executeID}
	var response wireTaskRunInfoResponse
	path := "/task/runs/" + url.PathEscape(executeID)
	if err := c.callBusiness(ctx, invocation, path, &request, &response, false, "get task run", nil); err != nil {
		return TaskRun{}, err
	}
	return taskRunFromWire(response.Data), nil
}

func (c *HTTPClient) ObserveTask(ctx context.Context, invocation Invocation, name string) (TaskObservation, error) {
	task, err := c.GetTask(ctx, invocation, name)
	if err != nil {
		return TaskObservation{}, err
	}
	// One page, per the Client.ObserveTask contract. Reading a task's whole
	// history on every poll would cost a request per hundred runs forever, to
	// re-derive a fact the caller already latched.
	runPage, err := c.listTaskRuns(
		ctx, invocation, task.ID, ListTaskRunsOptions{Page: 1, PerPage: listPageSize})
	if err != nil {
		return TaskObservation{}, err
	}
	runs := runPage.Runs
	observation := TaskObservation{
		TaskName:       task.Name,
		TaskStatus:     task.Status,
		LastResolvedTS: task.LastProgress.LastResolvedTS,
		Progress:       task.LastProgress,
		ObservedAt:     c.clock().UTC(),
	}
	if len(runs) > 0 {
		latest := latestRun(runs)
		observation.LatestRunID = latest.ExecuteID
		observation.LatestRunStatus = latest.Status
		observation.LatestRunMessage = latest.Message
		// A SnapshotAndCdc task keeps two runs alive at once and writes a
		// disjoint half of the progress to each: the dumpling run carries
		// fullSyncCompleted and backupTSO, the CDC run carries lastResolvedTS
		// and the row counters. Reading only the latest run would therefore
		// never see full-sync completion once the CDC run exists.
		//
		// /task/runs also returns historical runs, so only the facts that are
		// cumulative properties of the task are merged across all of them. The
		// per-run counters are taken from the newest run alone - summing or
		// maxing them over history describes no real quantity, and a
		// long-finished run with tablesError would otherwise pin the task to
		// "erroring" forever.
		mergeRunCounters(&observation.Progress, latest.Progress)
		for _, run := range runs {
			mergeTaskFacts(&observation.Progress, run.Progress)
		}
		if observation.Progress.LastResolvedTS > observation.LastResolvedTS {
			observation.LastResolvedTS = observation.Progress.LastResolvedTS
		}
	}
	observation.LastResolvedAt = tsoTime(observation.LastResolvedTS)
	return observation, nil
}

func notFoundError(what string) error {
	return &APIError{StatusCode: http.StatusNotFound, Code: ErrorCodeNotFound, Message: what + " does not exist"}
}

func (c *HTTPClient) buildDataSourceRequest(dataSourceID int32, spec DataSourceSpec) (wireDataSourceRequest, []string, error) {
	if strings.TrimSpace(spec.Name) == "" {
		return wireDataSourceRequest{}, nil, errors.New("datasource name is required")
	}
	if strings.TrimSpace(spec.S3.Bucket) == "" || strings.TrimSpace(spec.S3.Region) == "" {
		return wireDataSourceRequest{}, nil, errors.New("S3 bucket and region are required")
	}
	usesRole := strings.TrimSpace(spec.S3.RoleARN) != ""
	usesKeys := !spec.S3.AccessKey.Empty() && !spec.S3.SecretKey.Empty()
	if !usesRole && !usesKeys {
		return wireDataSourceRequest{}, nil, errors.New("either role ARN or S3 access key and secret key are required")
	}
	request := wireDataSourceRequest{
		DataSourceID: dataSourceID,
		Name:         spec.Name,
		ServiceID:    ServiceIDTiDBCDC,
		TiDBCredentials: wireS3Credentials{
			S3Bucket:    spec.S3.Bucket,
			S3Region:    spec.S3.Region,
			S3Endpoint:  spec.S3.Endpoint,
			S3AccessKey: spec.S3.AccessKey.value,
			S3SecretKey: spec.S3.SecretKey.value,
			SQSQueueURL: spec.S3.SQSQueueURL,
			SQSRegion:   spec.S3.SQSRegion,
			RoleARN:     spec.S3.RoleARN,
			ExternalID:  spec.S3.ExternalID.value,
		},
	}
	secrets := []string{spec.S3.AccessKey.value, spec.S3.SecretKey.value, spec.S3.ExternalID.value}
	return request, secrets, nil
}

// validateTaskSpecForWire covers everything checkable without the wire's
// numeric datasource id.
func validateTaskSpecForWire(spec TaskSpec) error {
	if err := validateTableRules(spec.TiDB.TableRules); err != nil {
		return err
	}
	if strings.TrimSpace(spec.Name) == "" || strings.TrimSpace(spec.DataSourceName) == "" {
		return errors.New("task name and datasource are required")
	}
	if strings.TrimSpace(spec.Warehouse) == "" {
		return errors.New("warehouse is required")
	}
	return validateTaskSpec(spec.TiDB)
}

func (c *HTTPClient) buildTaskRequest(taskID, dataSourceID int32, spec TaskSpec) (wireTaskRequest, error) {
	tableRules := append([]string(nil), spec.TiDB.TableRules...)
	if strings.TrimSpace(spec.Name) == "" || dataSourceID <= 0 {
		return wireTaskRequest{}, errors.New("task name and datasource are required")
	}
	if strings.TrimSpace(spec.Warehouse) == "" {
		return wireTaskRequest{}, errors.New("warehouse is required")
	}
	if err := validateTaskSpec(spec.TiDB); err != nil {
		return wireTaskRequest{}, err
	}
	continuousIngestion := int32(0)
	if spec.ContinuousIngestion {
		continuousIngestion = 1
	}
	return wireTaskRequest{
		TaskID:              taskID,
		Name:                spec.Name,
		DataSourceID:        dataSourceID,
		Warehouse:           spec.Warehouse,
		ContinuousIngestion: continuousIngestion,
		TiDBTask: wireTiDBTask{
			SyncMode:                wireSyncMode(spec.TiDB.SyncMode),
			TableRules:              tableRules,
			PollIntervalSeconds:     spec.TiDB.PollIntervalSeconds,
			BatchFileCount:          spec.TiDB.BatchFileCount,
			MergeIntervalSeconds:    spec.TiDB.MergeIntervalSeconds,
			TableParallelism:        spec.TiDB.TableParallelism,
			AllowDelete:             spec.TiDB.AllowDelete,
			DumplingS3Prefix:        spec.TiDB.DumplingS3Prefix,
			DumplingCSVSeparator:    spec.TiDB.DumplingCSVSeparator,
			DumplingCSVNull:         spec.TiDB.DumplingCSVNull,
			DumplingCSVSkipHeader:   spec.TiDB.DumplingCSVSkipHeader,
			DumplingAutoCreateTable: spec.TiDB.DumplingAutoCreateTable,
			DumplingOnError:         spec.TiDB.DumplingOnError,
			DumplingPurge:           spec.TiDB.DumplingPurge,
			ConflictKey:             spec.TiDB.ConflictKey,
			ChangefeedS3Prefix:      spec.TiDB.ChangefeedS3Prefix,
			TargetDatabasePrefix:    spec.TiDB.TargetNaming.DatabasePrefix,
			TargetDatabaseSuffix:    spec.TiDB.TargetNaming.DatabaseSuffix,
			TargetTablePrefix:       spec.TiDB.TargetNaming.TablePrefix,
			TargetTableSuffix:       spec.TiDB.TargetNaming.TableSuffix,
		},
	}, nil
}

// validateTableRules keeps a task from being created with nothing to match.
// Lake would accept an empty rule list and then discover every database under
// the prefix, which is never what a caller with a filter meant.
func validateTableRules(rules TableRules) error {
	if len(rules) == 0 {
		return errors.New("at least one table rule is required")
	}
	for _, rule := range rules {
		if strings.TrimSpace(rule) == "" {
			return errors.New("table rules must not be blank")
		}
	}
	return nil
}

func validateTaskSpec(spec TiDBTaskSpec) error {
	switch spec.SyncMode {
	case SyncModeSnapshot:
		if spec.DumplingS3Prefix == "" {
			return errors.New("dumpling S3 prefix is required for Snapshot mode")
		}
	case SyncModeCDC:
		if spec.ChangefeedS3Prefix == "" {
			return errors.New("changefeed S3 prefix is required for CDC mode")
		}
	case SyncModeSnapshotAndCDC:
		if spec.DumplingS3Prefix == "" || spec.ChangefeedS3Prefix == "" {
			return errors.New("dumpling and changefeed S3 prefixes are required for SnapshotAndCdc mode")
		}
	default:
		return errors.New("supported TiDB sync mode is required")
	}
	values := []struct {
		name  string
		value *int32
	}{
		{name: "poll interval", value: spec.PollIntervalSeconds},
		{name: "batch file count", value: spec.BatchFileCount},
		{name: "merge interval", value: spec.MergeIntervalSeconds},
		{name: "table parallelism", value: spec.TableParallelism},
		{name: "CSV skip header", value: spec.DumplingCSVSkipHeader},
	}
	for _, candidate := range values {
		if candidate.value != nil && *candidate.value < 0 {
			return fmt.Errorf("%s must not be negative", candidate.name)
		}
	}
	return nil
}

func (c *HTTPClient) callBusiness(
	ctx context.Context,
	invocation Invocation,
	path string,
	requestBody orgScopedRequest,
	responseBody any,
	write bool,
	operation string,
	requestSecrets []string,
) error {
	if err := invocation.validate(); err != nil {
		return err
	}
	identity, err := c.exchangeToken(ctx, invocation)
	if err != nil {
		return err
	}
	requestBody.setOrgSlug(identity.orgSlug)
	body, err := json.Marshal(requestBody)
	if err != nil {
		return fmt.Errorf("marshal Lake %s request: %w", operation, err)
	}
	if len(body) > maxBusinessRequestBytes {
		return errors.New("Lake business request exceeds 1 MiB")
	}
	requestID, err := c.requestID()
	if err != nil {
		return fmt.Errorf("generate Lake request id: %w", err)
	}
	fullPath := businessPathPrefix + path
	request, err := http.NewRequestWithContext(ctx, http.MethodPost, c.baseURL+fullPath, bytes.NewReader(body))
	if err != nil {
		return fmt.Errorf("build Lake %s request: %w", operation, err)
	}
	request.Header.Set("Authorization", "Bearer "+identity.accessToken)
	request.Header.Set("Content-Type", "application/json")
	request.Header.Set("X-Request-ID", requestID)
	if invocation.hasContext() {
		request.Header.Set("X-TiDB-Context-Type", invocation.ContextType)
		request.Header.Set("X-TiDB-Context-Id", invocation.ContextID)
		request.Header.Set("X-TiDB-Context-Version", strconv.FormatUint(invocation.ContextVersion, 10))
	}
	response, err := c.httpClient.Do(request)
	if err != nil {
		if write {
			return &UnknownWriteResultError{Operation: operation, Err: err}
		}
		return fmt.Errorf("call Lake %s: %w", operation, err)
	}
	defer response.Body.Close() //nolint:errcheck
	responseBytes, readErr := readLimited(response.Body, maxBusinessResponseBytes)
	if readErr != nil {
		if write {
			return &UnknownWriteResultError{Operation: operation, Err: readErr}
		}
		return fmt.Errorf("read Lake %s response: %w", operation, readErr)
	}
	// The organization slug is deliberately absent: it identifies the tenant
	// rather than authenticating it, and it is handed back to callers as the
	// external ID, so redacting it here would only cost debuggability.
	secrets := append([]string{c.hmacSecret.value, identity.accessToken}, requestSecrets...)
	if response.StatusCode < http.StatusOK || response.StatusCode >= http.StatusMultipleChoices {
		apiErr := apiErrorFromResponse(response.StatusCode, responseBytes, secrets)
		if write && (response.StatusCode == http.StatusRequestTimeout || response.StatusCode >= http.StatusInternalServerError) {
			return &UnknownWriteResultError{Operation: operation, Err: apiErr}
		}
		return apiErr
	}
	var envelope wireSuccessResponse
	if err := json.Unmarshal(responseBytes, &envelope); err != nil {
		if write {
			return &UnknownWriteResultError{Operation: operation, Err: errors.New("invalid Lake success response")}
		}
		return errors.New("invalid Lake success response")
	}
	if !envelope.Success {
		// Lake can report a business failure inside a 200. Parse the body the
		// same way as a transport error so the error code survives and
		// not-found stays distinguishable.
		return apiErrorFromResponse(response.StatusCode, responseBytes, secrets)
	}
	if err := json.Unmarshal(responseBytes, responseBody); err != nil {
		// Carry the decode error. It names the offending field and types,
		// which is the only way to tell a Lake schema change apart from a
		// mistake on this side; the body itself stays out of it because it can
		// hold credentials.
		payloadErr := fmt.Errorf("invalid Lake success response payload: %w", err)
		if write {
			return &UnknownWriteResultError{Operation: operation, Err: payloadErr}
		}
		return payloadErr
	}
	return nil
}

type exchangedIdentity struct {
	accessToken string
	orgSlug     string
}

func (c *HTTPClient) exchangeToken(ctx context.Context, invocation Invocation) (exchangedIdentity, error) {
	if err := invocation.validate(); err != nil {
		return exchangedIdentity{}, err
	}
	body, err := json.Marshal(tokenExchangeRequest{
		TiDBOrgID:      invocation.TiDBOrgID,
		LakeOrgID:      invocation.LakeOrgID,
		OrgSlug:        invocation.OrgSlug,
		ContextType:    invocation.ContextType,
		ContextID:      invocation.ContextID,
		ContextVersion: invocation.ContextVersion,
		InitiatedBy:    invocation.InitiatedBy,
	})
	if err != nil {
		return exchangedIdentity{}, fmt.Errorf("marshal Lake token exchange request: %w", err)
	}
	nonce, err := c.nonce()
	if err != nil {
		return exchangedIdentity{}, fmt.Errorf("generate Lake token exchange nonce: %w", err)
	}
	timestamp := c.clock().Unix()
	signature := sign(c.hmacSecret.value, http.MethodPost, tokenExchangePath, timestamp, nonce, body)
	request, err := http.NewRequestWithContext(ctx, http.MethodPost, c.baseURL+tokenExchangePath, bytes.NewReader(body))
	if err != nil {
		return exchangedIdentity{}, fmt.Errorf("build Lake token exchange request: %w", err)
	}
	request.Header.Set("Content-Type", "application/json")
	request.Header.Set("X-Trusted-Timestamp", strconv.FormatInt(timestamp, 10))
	request.Header.Set("X-Trusted-Nonce", nonce)
	request.Header.Set("X-Trusted-Signature", signature)
	response, err := c.httpClient.Do(request)
	if err != nil {
		return exchangedIdentity{}, fmt.Errorf("call Lake token exchange: %w", err)
	}
	defer response.Body.Close() //nolint:errcheck
	responseBytes, err := readLimited(response.Body, maxTokenResponseBytes)
	if err != nil {
		return exchangedIdentity{}, fmt.Errorf("read Lake token exchange response: %w", err)
	}
	if response.StatusCode < http.StatusOK || response.StatusCode >= http.StatusMultipleChoices {
		return exchangedIdentity{}, apiErrorFromResponse(response.StatusCode, responseBytes, []string{c.hmacSecret.value})
	}
	var exchangeResponse tokenExchangeResponse
	if err := json.Unmarshal(responseBytes, &exchangeResponse); err != nil {
		return exchangedIdentity{}, errors.New("invalid Lake token exchange response")
	}
	if !exchangeResponse.Success || exchangeResponse.Data.AccessToken == "" {
		return exchangedIdentity{}, errors.New("Lake token exchange response has no access token")
	}
	if strings.TrimSpace(exchangeResponse.Data.OrgSlug) == "" {
		return exchangedIdentity{}, errors.New("Lake token exchange response has no organization slug")
	}
	if invocation.OrgSlug != "" && exchangeResponse.Data.OrgSlug != invocation.OrgSlug {
		return exchangedIdentity{}, errors.New("Lake token exchange returned a different organization")
	}
	return exchangedIdentity{
		accessToken: exchangeResponse.Data.AccessToken,
		orgSlug:     exchangeResponse.Data.OrgSlug,
	}, nil
}

func sign(secret, method, path string, timestamp int64, nonce string, body []byte) string {
	bodyHash := sha256.Sum256(body)
	canonical := fmt.Sprintf("%s\n%s\n%d\n%s\n%x", strings.ToUpper(method), path, timestamp, nonce, bodyHash)
	mac := hmac.New(sha256.New, []byte(secret))
	_, _ = mac.Write([]byte(canonical))
	return hex.EncodeToString(mac.Sum(nil))
}

func randomUUID() (string, error) {
	value := make([]byte, 16)
	if _, err := rand.Read(value); err != nil {
		return "", err
	}
	value[6] = (value[6] & 0x0f) | 0x40
	value[8] = (value[8] & 0x3f) | 0x80
	return fmt.Sprintf("%x-%x-%x-%x-%x", value[0:4], value[4:6], value[6:8], value[8:10], value[10:16]), nil
}

func readLimited(reader io.Reader, limit int64) ([]byte, error) {
	limited := io.LimitReader(reader, limit+1)
	data, err := io.ReadAll(limited)
	if err != nil {
		return nil, err
	}
	if int64(len(data)) > limit {
		return nil, errors.New("Lake response exceeds size limit")
	}
	return data, nil
}

func apiErrorFromResponse(statusCode int, body []byte, secrets []string) error {
	var response wireErrorResponse
	_ = json.Unmarshal(body, &response)
	return &APIError{
		StatusCode: statusCode,
		Code:       response.Error,
		Message:    sanitizeMessage(response.Message, secrets),
	}
}

func sanitizeMessage(message string, secrets []string) string {
	const maxMessageLength = 2048
	for _, secret := range secrets {
		if secret != "" {
			message = strings.ReplaceAll(message, secret, "[REDACTED]")
		}
	}
	if len(message) > maxMessageLength {
		message = message[:maxMessageLength]
	}
	return message
}

func dataSourceFromWire(item wireDataSourceItem) (DataSource, error) {
	credentials, err := credentialsFromServiceArgs(item.ServiceArgs)
	if err != nil {
		return DataSource{}, errors.New("invalid Lake datasource service arguments")
	}
	id := item.ID
	if id == 0 {
		id = item.DataSourceID
	}
	return DataSource{
		ID:   id,
		Name: item.Name,
		S3: S3Credentials{
			Bucket:   credentials.S3Bucket,
			Region:   credentials.S3Region,
			Endpoint: credentials.S3Endpoint,
			RoleARN:  credentials.RoleARN,
			// serviceArgs is the credential snapshot returned verbatim, so
			// these come back exactly as they were sent. Callers compare them
			// to decide whether a datasource still matches their spec, and a
			// field dropped here would read as a permanent mismatch in one
			// direction and a silently skipped update in the other.
			SQSQueueURL: credentials.SQSQueueURL,
			SQSRegion:   credentials.SQSRegion,
		},
		CreatedAt: parseTime(item.CreatedAt),
		UpdatedAt: parseTime(item.UpdatedAt),
	}, nil
}

func credentialsFromServiceArgs(raw json.RawMessage) (wireS3Credentials, error) {
	if len(raw) == 0 || bytes.Equal(bytes.TrimSpace(raw), []byte("null")) {
		return wireS3Credentials{}, nil
	}
	payload := raw
	if bytes.HasPrefix(bytes.TrimSpace(raw), []byte{'"'}) {
		var encoded string
		if err := json.Unmarshal(raw, &encoded); err != nil {
			return wireS3Credentials{}, err
		}
		payload = []byte(encoded)
	}
	var credentials wireS3Credentials
	if err := json.Unmarshal(payload, &credentials); err != nil {
		return wireS3Credentials{}, err
	}
	return credentials, nil
}

func taskFromWire(item wireTaskItem) Task {
	id := item.ID
	if id == 0 {
		id = item.TaskID
	}
	continuous := item.ContinuousIngestion == 1
	return Task{
		ID:                  id,
		Name:                item.Name,
		DataSourceID:        item.DataSourceID,
		Warehouse:           item.Warehouse,
		ContinuousIngestion: continuous,
		Status:              TaskStatus(item.TaskStatus),
		Database:            item.Database,
		TiDB:                taskSpecFromWire(item.TiDBTask),
		LastProgress:        progressFromWire(item.LastRunData.TiDBLastRun),
		CreatedAt:           parseTime(item.CreatedAt),
		UpdatedAt:           parseTime(item.UpdatedAt),
	}
}

func taskSpecFromWire(value wireTiDBTask) TiDBTaskSpec {
	return TiDBTaskSpec{
		SyncMode:                SyncMode(value.SyncMode),
		TableRules:              append(TableRules(nil), value.TableRules...),
		PollIntervalSeconds:     value.PollIntervalSeconds,
		BatchFileCount:          value.BatchFileCount,
		MergeIntervalSeconds:    value.MergeIntervalSeconds,
		TableParallelism:        value.TableParallelism,
		AllowDelete:             value.AllowDelete,
		DumplingS3Prefix:        value.DumplingS3Prefix,
		DumplingCSVSeparator:    value.DumplingCSVSeparator,
		DumplingCSVNull:         value.DumplingCSVNull,
		DumplingCSVSkipHeader:   value.DumplingCSVSkipHeader,
		DumplingAutoCreateTable: value.DumplingAutoCreateTable,
		DumplingOnError:         value.DumplingOnError,
		DumplingPurge:           value.DumplingPurge,
		ConflictKey:             value.ConflictKey,
		ChangefeedS3Prefix:      value.ChangefeedS3Prefix,
		TargetNaming: TargetNaming{
			DatabasePrefix: value.TargetDatabasePrefix,
			DatabaseSuffix: value.TargetDatabaseSuffix,
			TablePrefix:    value.TargetTablePrefix,
			TableSuffix:    value.TargetTableSuffix,
		},
	}
}

// runMessageFromWire prefers the per-run message over the envelope's error:
// both carry the same text on the runs seen so far, and the inner one is the
// field Lake documents as the run's own outcome.
func runMessageFromWire(data wireResponseData) string {
	if message := strings.TrimSpace(data.TiDBLastRun.LastMessage); message != "" {
		return message
	}
	return strings.TrimSpace(data.Error)
}

func taskRunFromWire(item wireTaskRunItem) TaskRun {
	// requestData echoes the request that created the run, which for a TiDB CDC
	// task contains the S3 credentials. It is dropped here for the same reason
	// serviceArgs is: nothing outside this package may see it.
	return TaskRun{
		ExecuteID: item.ExecuteID,
		TaskID:    item.TaskID,
		Status:    TaskRunStatus(item.Status),
		Message:   runMessageFromWire(item.ResponseData),
		Progress:  progressFromWire(item.ResponseData.TiDBLastRun),
		CreatedAt: parseTime(item.CreatedAt),
		UpdatedAt: parseTime(item.UpdatedAt),
	}
}

// mergeRunCounters takes the counters that describe a single run. They are only
// meaningful for one execution, so they come from that run and are not combined
// with any other.
func mergeRunCounters(into *TiDBProgress, from TiDBProgress) {
	if from.RowsSynced == 0 && from.TablesActive == 0 && from.TablesError == 0 &&
		from.DatabasesActive == 0 {
		// The dumpling run reports none of these; leave whatever the task's own
		// lastRunData supplied rather than blanking it.
		return
	}
	into.RowsSynced = from.RowsSynced
	into.InsertCount = from.InsertCount
	into.UpdateCount = from.UpdateCount
	into.DeleteCount = from.DeleteCount
	into.TablesActive = from.TablesActive
	into.TablesError = from.TablesError
	into.DatabasesActive = from.DatabasesActive
}

// mergeTaskFacts folds in the values that belong to the task rather than to one
// run, so they are safe to take from any run that reported them: the resolved
// position only moves forward, and a full sync that completed stays completed.
func mergeTaskFacts(into *TiDBProgress, from TiDBProgress) {
	if from.LastResolvedTS > into.LastResolvedTS {
		into.LastResolvedTS = from.LastResolvedTS
	}
	if from.SyncMode != "" {
		into.SyncMode = from.SyncMode
	}
	if from.FullSyncCompleted {
		into.FullSyncCompleted = true
		if !from.FullSyncCompletedAt.IsZero() {
			into.FullSyncCompletedAt = from.FullSyncCompletedAt
		}
	}
	if from.BackupTSO > into.BackupTSO {
		into.BackupTSO = from.BackupTSO
	}
}

// newestRuns is the window an observation reads. The fake applies it too, so a
// fake-backed test cannot depend on a run the real client would have paged
// past.
func newestRuns(runs []TaskRun) []TaskRun {
	if len(runs) <= listPageSize {
		return runs
	}
	sorted := append([]TaskRun(nil), runs...)
	sort.SliceStable(sorted, func(i, j int) bool {
		return sorted[i].CreatedAt.After(sorted[j].CreatedAt)
	})
	return sorted[:listPageSize]
}

func latestRun(runs []TaskRun) TaskRun {
	latest := runs[0]
	for _, run := range runs[1:] {
		if !run.CreatedAt.IsZero() && (latest.CreatedAt.IsZero() || run.CreatedAt.After(latest.CreatedAt)) {
			latest = run
		}
	}
	return latest
}

func tsoTime(tso uint64) time.Time {
	if tso == 0 {
		return time.Time{}
	}
	physicalMillis := tso >> 18
	if physicalMillis > uint64(^uint64(0)>>1) {
		return time.Time{}
	}
	return time.UnixMilli(int64(physicalMillis)).UTC()
}