internal/api

import "github.com/neochaotic/leoflow/internal/api"

const DefaultUIAutoRefreshIntervalSeconds = 30

var ErrNotFound = domain.ErrNotFound

func AbortProblem(c *gin.Context, status int, title, detail string)

func CORS(allowed []string) gin.HandlerFunc

func DevBypassAuth() gin.HandlerFunc

func JWTAuth(authn auth.Authenticator) gin.HandlerFunc

func NewServer(deps Dependencies) *gin.Engine

func NoStoreOnVolatileRoutes() gin.HandlerFunc

func ObservabilityHandler(registry *prometheus.Registry, checks map[string]HealthChecker) http.Handler

func Observe(metrics Metrics, tracer trace.Tracer) gin.HandlerFunc

func RequestID() gin.HandlerFunc

func RequirePermission(action, resource string) gin.HandlerFunc

func StructuredLogger(logger *slog.Logger) gin.HandlerFunc

func UserFromContext(c *gin.Context) (*auth.User, bool)

type AuditLogReader interface {
    ListAuditLogs(ctx context.Context, tenant, dagID string, limit, offset int) ([]domain.AuditLogEntry, int, error)
}

type AuditWriter interface {
    RecordTaskActionAudit(ctx context.Context, tenant, userID, action, dagID, runID, taskID string, tryNumber int) error
}

type AuthAuditWriter interface {
    RecordAuthEvent(ctx context.Context, tenant, actorUserID, action, email, outcome string, extra map[string]string) error
}

type ConnectionStore interface {
    ListConnections(ctx context.Context, tenant string, limit, offset int) ([]domain.Connection, int, error)
    GetConnection(ctx context.Context, tenant, connID string) (domain.Connection, error)
    SetConnection(ctx context.Context, tenant string, c domain.Connection) error
    DeleteConnection(ctx context.Context, tenant, connID string) error
}

type ConnectionTester interface {
    Test(ctx context.Context, c domain.Connection) (ok bool, message string)
}

type DagLatestRunsReader interface {
    LatestRunsForDags(ctx context.Context, tenant string, dagIDs []string, perDag int) (map[string][]domain.DagRun, error)
}

type DagRepository interface {
    ListDags(ctx context.Context, tenant string, limit, offset int) ([]domain.DAG, int, error)
    GetDag(ctx context.Context, tenant, dagID string) (domain.DAG, error)
    SetPaused(ctx context.Context, tenant, dagID string, paused bool) (domain.DAG, error)
    DeleteDag(ctx context.Context, tenant, dagID string) error
    ClearDagHistory(ctx context.Context, tenant, dagID string) error
    ListDagsFiltered(ctx context.Context, tenant, runState string, paused *bool, limit, offset int) ([]domain.DAG, int, error)
}

type DagRunRepository interface {
    ListDagRuns(ctx context.Context, tenant, dagID string, limit, offset int) ([]domain.DagRun, int, error)
    GetDagRun(ctx context.Context, tenant, dagID, runID string) (domain.DagRun, error)
    CreateDagRun(ctx context.Context, tenant, dagID string, run domain.DagRun) (domain.DagRun, error)
    SetDagRunState(ctx context.Context, tenant, dagID, runID, state string) error
    DeleteDagRun(ctx context.Context, tenant, dagID, runID string) error
}

type DagSpecReader interface {
    GetCurrentSpec(ctx context.Context, tenant, dagID string) (domain.DAGSpec, error)
}

type DagVersionLister interface {
    ListDagVersions(ctx context.Context, tenant, dagID string) ([]domain.DagVersion, error)
}

type DagVersionRepository interface {
    // RegisterDagVersion upserts the DAG and inserts a version keyed by
    // specHash, reporting whether a new version was created (false if the hash
    // already existed — the push is idempotent).
    RegisterDagVersion(ctx context.Context, tenant string, spec domain.DAGSpec, specHash string) (bool, error)
}

type DashboardStatsReader interface {
    DagStats(ctx context.Context, tenant string) (domain.DagStats, error)
    HistoricalMetrics(ctx context.Context, tenant string, since, until time.Time) (domain.HistoricalMetrics, error)
}

type Dependencies struct {
    Logger        *slog.Logger
    Authenticator auth.Authenticator
    RateLimiter   *auth.RateLimiter
    Registry      *prometheus.Registry
    Metrics       Metrics
    Tracer        trace.Tracer
    HealthChecks  map[string]HealthChecker
    CORSOrigins   []string
    // TrustedProxies is the set of proxy IPs/CIDRs whose X-Forwarded-For header
    // gin will honor when resolving c.ClientIP(). Empty/nil trusts NO proxy, so
    // ClientIP is the direct peer and a spoofed XFF cannot forge the client IP
    // (audit H1). A Pro deployment behind an ingress sets this to the ingress
    // CIDR so per-client rate-limiting and audit see the real client.
    TrustedProxies []string
    TokenTTLSecs   int
    // TokenRenewer re-mints a still-valid user bearer with a fresh short TTL so a
    // long CLI/dev session need not re-login every TokenTTLSecs (aresta #5). Nil
    // leaves the renew route unregistered (renewal simply unavailable). In practice
    // it is the same *auth.JWTAuthenticator as Authenticator.
    TokenRenewer TokenRenewer
    // TokenMaxLifetimeSecs is the hard ceiling on a renewed session's total age
    // since first login; past it, renewal is refused and the user must
    // re-authenticate. Non-positive disables the ceiling.
    TokenMaxLifetimeSecs int
    // InstanceName is shown in the UI navbar (Airflow's instance_name). Empty
    // falls back to "Leoflow"; `leoflow dev` sets it to mark the DEV environment.
    InstanceName string
    // UIAutoRefreshIntervalSeconds controls the SPA's polling cadence for DAG /
    // DagRun / task-instance state refresh (Airflow's auto_refresh_interval).
    // Non-positive (the zero default) falls back to DefaultUIAutoRefreshIntervalSeconds
    // (30s, production-safe). `leoflow lite` sets it to ~5s for a snappy inner loop.
    UIAutoRefreshIntervalSeconds int
    // DevNoAuth replaces JWT auth with a dev-only bypass that authenticates every
    // request as an admin (no login). It is for `leoflow dev` only and must never
    // be set in production. See DevBypassAuth.
    DevNoAuth bool
    // Edition marks the running edition ("pro", "lite", or empty). It gates
    // Pro-only surfaces: named-pool CRUD is registered as real endpoints only when
    // Edition == "pro" (ADR 0053), otherwise the Pools screen gets the graceful
    // empty-collection stub, matching how the scheduler's pool gate is Pro-gated.
    Edition string

    // Resource repositories. Routes for nil repositories are not registered.
    Dags           DagRepository
    DagRuns        DagRunRepository
    Tasks          TaskInstanceRepository
    Versions       DagVersionRepository
    Xcoms          XComReader
    Logs           LogReader
    Specs          DagSpecReader
    LatestRuns     DagLatestRunsReader
    TaskSummary    TaskSummaryReader
    DagVersions    DagVersionLister
    DashboardStats DashboardStatsReader
    AuditLog       AuditLogReader
    Variables      VariableStore
    Users          UserStore
    UserAudit      UserAuditWriter
    Connections    ConnectionStore
    ConnectionTest ConnectionTester
    Pools          PoolStore
    Favorites      FavoriteStore
    ImportErrors   ImportErrorStore
    Audit          AuditWriter
    ExecutorInfo   ExecutorInfo

    // Workspace backs the Lite web editor (ADR 0025). When nil the editor's
    // filesystem API is not registered (Production, or Lite without a workspace).
    Workspace WorkspaceFS

    // MonacoDir is the directory holding the pinned Monaco bundle that
    // `leoflow setup` fetched; the editor page is served Monaco from it. Empty or
    // missing makes the page show a setup hint instead of a broken editor.
    MonacoDir string

    // ExamplesFS backs the IDE's "Download examples" button — typically the
    // `embed.FS` shipped from the leoflow root package. Nil disables the button.
    ExamplesFS fs.FS

    // SchedulerHealth reports the scheduler's heartbeat for /monitor/health.
    // When nil the component reports healthy (single-process role assumption).
    SchedulerHealth Heartbeater

    // UI serves the embedded SPA. When nil the server is API-only.
    UI  UIServer

    // OIDC wiring (registered only when OIDCFlow is non-nil — provider: oidc).
    // The JWT authenticator above stays the request-path verifier in both modes.
    //
    // OIDCFlow is the discovered Authorization Code + PKCE flow; nil in JWT mode,
    // in which case the /api/v2/auth/oidc/* routes are not registered.
    OIDCFlow *oidc.Flow
    // OIDCSettings carries the role mappings, JIT policy, default_role, and
    // break-glass allowlist the login flow and the credential gate read.
    OIDCSettings config.OIDCSection
    // OIDCUsers resolves and JIT-provisions OIDC identities (the storage repo).
    OIDCUsers OIDCUserStore
    // AuthAudit records authentication events (login, tenant-pin rejection, JIT,
    // break-glass, logout) to the audit sink.
    AuthAudit AuthAuditWriter
    // JWTSecret is the HS256 secret the OIDC callback mints the app's _token with.
    JWTSecret string
}

type ExecutorInfo struct {
    PodDispatchEnabled    bool
    TaskNamespace         string
    AgentControlPlaneAddr string
}

type FavoriteStore interface {
    AddFavorite(ctx context.Context, tenant, userID, dagID string) error
    RemoveFavorite(ctx context.Context, tenant, userID, dagID string) error
    FavoriteDagIDs(ctx context.Context, tenant, userID string) (map[string]bool, error)
}

type HealthChecker interface {
    Ping(ctx context.Context) error
}

type Heartbeater interface {
    Heartbeat() (healthy bool, last time.Time)
}

type ImportErrorStore interface {
    ListImportErrors(ctx context.Context, tenant string) ([]domain.ImportError, error)
    SetImportError(ctx context.Context, tenant string, e domain.ImportError) error
    ClearImportError(ctx context.Context, tenant, filename string) error
}

type LogReader interface {
    ReadLogs(ctx context.Context, tenant, dagID, runID, taskID string, tryNumber int) (io.ReadCloser, error)
    Tail(ctx context.Context, tenant, dagID, runID, taskID string, tryNumber int) (<-chan string, func(), error)
}

type Metrics interface {
    RecordHTTPRequest(method, path string, status int, dur time.Duration)
}

type OIDCUserStore interface {
    // FindUserByOIDCSubject resolves a returning identity by its (provider,
    // subject) pair, returning the user, whether it is active, and
    // auth.ErrUserNotFound when no row matches.
    FindUserByOIDCSubject(ctx context.Context, provider, subject string) (*auth.User, bool, error)
    // CreateOIDCUser JIT-provisions an OIDC-only user with the given roles.
    CreateOIDCUser(ctx context.Context, tenant, email, provider, subject string, roles []string) (*auth.User, error)
    // RoleExists reports whether a role name exists for the tenant, used to fail
    // closed on a misconfigured default_role or role mapping.
    RoleExists(ctx context.Context, tenant, role string) (bool, error)
    // ReconcileUserRoles sets the user's DB roles to EXACTLY roleNames (the IdP is
    // authoritative). It fails closed on a name that is not a role in the tenant,
    // leaving the prior grants intact.
    ReconcileUserRoles(ctx context.Context, userID string, roleNames []string) error
}

type PoolStore interface {
    ListPools(ctx context.Context, tenant string, limit, offset int) ([]domain.Pool, int, error)
    GetPool(ctx context.Context, tenant, name string) (domain.Pool, error)
    SetPool(ctx context.Context, tenant string, p domain.Pool) error
    DeletePool(ctx context.Context, tenant, name string) error
    PoolSlotUsage(ctx context.Context, tenant string) (map[string]domain.PoolUsage, error)
}

type Problem struct {
    Type     string `json:"type"`
    Title    string `json:"title"`
    Status   int    `json:"status"`
    Detail   string `json:"detail,omitempty"`
    Instance string `json:"instance,omitempty"`
}

type TaskInstanceRepository interface {
    ListTaskInstances(ctx context.Context, tenant, dagID, runID string, limit, offset int) ([]domain.TaskInstance, int, error)
    // ListTaskInstanceAttempts returns every attempt for (run, task), oldest
    // first — the current row UNIONed with the archived history. The UI's
    // /tries endpoint needs all attempts to render its navigable tabs.
    ListTaskInstanceAttempts(ctx context.Context, tenant, dagID, runID, taskID string) ([]domain.TaskInstance, error)
    ClearTaskInstances(ctx context.Context, tenant, dagID, runID string, taskIDs []string, onlyFailed, resetDagRun bool) (int, error)
    SetTaskInstanceState(ctx context.Context, tenant, dagID, runID, taskID, state string) error
}

type TaskSummaryReader interface {
    TaskInstancesForRuns(ctx context.Context, tenant, dagID string, runIDs []string) ([]domain.TaskInstance, error)
}

type TokenRenewer interface {
    RenewUserToken(token string, ttl, maxLifetime time.Duration) (renewed string, ok bool, err error)
}

type UIServer interface {
    StaticHandler() http.Handler
    Index(w http.ResponseWriter, basePath string)
}

type UserAuditWriter interface {
    RecordUserCreatedAudit(ctx context.Context, tenant, actorUserID, createdUserID, email, roles string) error
}

type UserStore interface {
    CreateUser(ctx context.Context, tenant, email, password string, roles []string) (domain.User, error)
    ListUsers(ctx context.Context, tenant string, limit, offset int) ([]domain.User, int, error)
}

type VariableStore interface {
    ListVariables(ctx context.Context, tenant string, limit, offset int) ([]domain.Variable, int, error)
    GetVariable(ctx context.Context, tenant, key string) (domain.Variable, error)
    SetVariable(ctx context.Context, tenant string, v domain.Variable) error
    DeleteVariable(ctx context.Context, tenant, key string) error
}

type WorkspaceFS interface {
    Tree() ([]workspace.Entry, error)
    Read(rel string) ([]byte, error)
    Write(rel string, data []byte) error
    Create(rel string, dir bool) error
    Move(from, to string) error
    Delete(rel string) error
}

type XComReader interface {
    GetXCom(ctx context.Context, tenant, dagID, runID, taskID, key string) (xcom.Entry, error)
    ListXComEntries(ctx context.Context, tenant, dagID, runID, taskID string) ([]domain.XComEntryMeta, error)
}