package connector_syncstate

import (
	"context"
)

// errorMessageMaxRunes is the maximum rune count stored in LastErrorMessage.
// Worker-supplied error text is truncated to this limit before storage so that
// upstream response bodies, tenant data, or secret-bearing diagnostics cannot
// leak verbatim through /api/connectors/:namespace/health to any caller with
// ViewConnectorHealth.
const errorMessageMaxRunes = 256

// truncateErrorMessage returns s truncated to errorMessageMaxRunes runes,
// with "…" appended when truncation occurred. The operation is rune-safe.
func truncateErrorMessage(s string) string {
	runes := []rune(s)
	if len(runes) <= errorMessageMaxRunes {
		return s
	}
	return string(runes[:errorMessageMaxRunes]) + "…"
}

// PackResolver verifies a connector-pack is active before SyncState reads/writes
// touch its rows. The controlplane adapter (backed by packs.ResolveConnectorPackByNamespace)
// satisfies this; mirrored from the namedcredential package's PackResolver shape.
type PackResolver interface {
	// ResolveConnectorPackByNamespace returns the per-incarnation InstallID and
	// whether the pack is currently active. Returns ("", false, nil) when not found.
	ResolveConnectorPackByNamespace(ctx context.Context, namespace string) (installID string, active bool, err error)
}

// Service is the worker-facing SyncState API. All methods gate on the connector
// pack being active; inactive/missing namespaces return ErrConnectorPackInactive.
// All methods are SERVER-INTERNAL — worker callers do not pass through any HTTP
// authz gate. The HTTP health endpoint enforces ViewConnectorHealth separately.
type Service struct {
	repo  repository
	packs PackResolver
}

func NewService(repo repository, packs PackResolver) *Service {
	return &Service{repo: repo, packs: packs}
}

// GetSyncState returns the row for (namespace, object, direction), or ErrNotFound.
func (s *Service) GetSyncState(ctx context.Context, namespace, object, direction string) (*SyncState, error) {
	if err := validateDirection(direction); err != nil {
		return nil, err
	}
	installID, err := s.gatePackActive(ctx, namespace)
	if err != nil {
		return nil, err
	}
	row, err := s.repo.Get(ctx, installID, object, direction)
	if err != nil {
		return nil, err
	}
	if row == nil {
		return nil, ErrNotFound
	}
	return row, nil
}

// UpdateCursor writes/refreshes Cursor + LastSyncAt for the tuple.
// Concurrent calls for the same tuple resolve as last-write-wins at the DB
// level via the unique (InstalledPackId, Object, Direction) index; no torn rows.
func (s *Service) UpdateCursor(ctx context.Context, namespace, object, direction, cursor string) error {
	if err := validateDirection(direction); err != nil {
		return err
	}
	installID, err := s.gatePackActive(ctx, namespace)
	if err != nil {
		return err
	}
	return s.repo.UpsertCursor(ctx, installID, namespace, object, direction, cursor)
}

// RecordError increments RunningErrorCount + sets LastErrorMessage/LastErrorAt.
// On a fresh row, RunningErrorCount initializes to 1.
//
// errorMsg is truncated to errorMessageMaxRunes runes (rune-safe, "…" suffix on
// truncation) before storage. This bounds the raw-error-text leak surface: the
// health endpoint surfaces the already-truncated value, so upstream response
// bodies, tenant data, or secret-bearing diagnostics cannot reach callers verbatim.
func (s *Service) RecordError(ctx context.Context, namespace, object, direction, errorMsg string) error {
	if err := validateDirection(direction); err != nil {
		return err
	}
	installID, err := s.gatePackActive(ctx, namespace)
	if err != nil {
		return err
	}
	errorMsg = truncateErrorMessage(errorMsg)
	return s.repo.UpsertError(ctx, installID, namespace, object, direction, errorMsg)
}

// ResetErrors clears the error-tracking columns. Returns ErrNotFound when no
// row exists; callers can detect mistargeted resets that way.
func (s *Service) ResetErrors(ctx context.Context, namespace, object, direction string) error {
	if err := validateDirection(direction); err != nil {
		return err
	}
	installID, err := s.gatePackActive(ctx, namespace)
	if err != nil {
		return err
	}
	rows, err := s.repo.ResetErrors(ctx, installID, object, direction)
	if err != nil {
		return err
	}
	if rows == 0 {
		return ErrNotFound
	}
	return nil
}

// ListByNamespace returns all SyncState rows for the given namespace.
// Gated on the connector pack being active.
func (s *Service) ListByNamespace(ctx context.Context, namespace string) ([]SyncState, error) {
	installID, err := s.gatePackActive(ctx, namespace)
	if err != nil {
		return nil, err
	}
	return s.repo.ListByInstallID(ctx, installID)
}

// gatePackActive verifies the named connector pack is active and returns its
// per-incarnation InstallID. Returns ErrConnectorPackInactive when inactive/unknown.
func (s *Service) gatePackActive(ctx context.Context, namespace string) (string, error) {
	installID, active, err := s.packs.ResolveConnectorPackByNamespace(ctx, namespace)
	if err != nil {
		return "", err
	}
	if !active {
		return "", ErrConnectorPackInactive
	}
	return installID, nil
}

func validateDirection(d string) error {
	switch d {
	case DirectionInbound, DirectionOutbound, DirectionBidirectional:
		return nil
	}
	return ErrInvalidDirection
}