diff --git a/config.example.yaml b/config.example.yaml index a16e502..e1647d2 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -72,3 +72,38 @@ storage: driver: "localfs" localfs: base_path: "./data/uploads" + +# 后台 job runner 配置 +jobs: + enabled: true + workers: 2 + shutdown_grace: 30s + workspace_fetch: + enabled: true + interval: 5m + worktree_prune: + enabled: true + interval: 24h + idle_threshold: 720h + warning_lead: 72h + attachment_cleanup: + enabled: true + interval: 24h + retention: 720h + job_reaper: + enabled: true + interval: 1m + lease_timeout: 5m + job_purge: + enabled: true + interval: 24h + completed_retention: 168h + +# 通知通道配置 +notify: + webhook: + enabled: false + url: "" + secret: "" + timeout: 10s + topics: [] diff --git a/internal/config/config.go b/internal/config/config.go index a29d51d..ecfbc60 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -39,6 +39,8 @@ type Config struct { Git Git `mapstructure:"git"` Chat Chat `mapstructure:"chat"` Storage Storage `mapstructure:"storage"` + Jobs JobsConfig `mapstructure:"jobs"` + Notify NotifyConfig `mapstructure:"notify"` } // HTTPConfig 描述 HTTP 服务监听参数。 @@ -107,6 +109,62 @@ type StorageLocalFS struct { BasePath string `mapstructure:"base_path"` } +// JobsConfig 控制后台 job runner 的总开关、worker 池与各定时任务参数。 +type JobsConfig struct { + Enabled bool `mapstructure:"enabled"` + Workers int `mapstructure:"workers"` + ShutdownGrace time.Duration `mapstructure:"shutdown_grace"` + WorkspaceFetch RunnerCfg `mapstructure:"workspace_fetch"` + WorktreePrune WorktreePruneCfg `mapstructure:"worktree_prune"` + AttachmentCleanup AttachmentCleanupCfg `mapstructure:"attachment_cleanup"` + JobReaper JobReaperCfg `mapstructure:"job_reaper"` + JobPurge JobPurgeCfg `mapstructure:"job_purge"` +} + +type RunnerCfg struct { + Enabled bool `mapstructure:"enabled"` + Interval time.Duration `mapstructure:"interval"` +} + +type WorktreePruneCfg struct { + Enabled bool `mapstructure:"enabled"` + Interval time.Duration `mapstructure:"interval"` + IdleThreshold time.Duration `mapstructure:"idle_threshold"` + WarningLead time.Duration `mapstructure:"warning_lead"` +} + +type AttachmentCleanupCfg struct { + Enabled bool `mapstructure:"enabled"` + Interval time.Duration `mapstructure:"interval"` + Retention time.Duration `mapstructure:"retention"` +} + +type JobReaperCfg struct { + Enabled bool `mapstructure:"enabled"` + Interval time.Duration `mapstructure:"interval"` + LeaseTimeout time.Duration `mapstructure:"lease_timeout"` +} + +type JobPurgeCfg struct { + Enabled bool `mapstructure:"enabled"` + Interval time.Duration `mapstructure:"interval"` + CompletedRetention time.Duration `mapstructure:"completed_retention"` +} + +// NotifyConfig 是 notify 模块的可选外发通道配置。当前仅 webhook。 +type NotifyConfig struct { + Webhook WebhookCfg `mapstructure:"webhook"` +} + +// WebhookCfg 是 webhook 通道的配置。enabled=false 时上层 dispatcher 仍会丢弃所有 webhook 消息。 +type WebhookCfg struct { + Enabled bool `mapstructure:"enabled"` + URL string `mapstructure:"url"` + Secret string `mapstructure:"secret"` + Timeout time.Duration `mapstructure:"timeout"` + Topics []string `mapstructure:"topics"` +} + // Load 从（按优先级递增）默认值 → 配置文件（如有）→ 环境变量 加载配置。 // configFile 可为空字符串，仅用环境变量。 func Load(configFile string) (*Config, error) { @@ -132,6 +190,26 @@ func Load(configFile string) (*Config, error) { v.SetDefault("chat.history.safety_margin_pct", 5.0) v.SetDefault("storage.driver", "localfs") v.SetDefault("storage.localfs.base_path", "./data/uploads") + v.SetDefault("jobs.enabled", true) + v.SetDefault("jobs.workers", 2) + v.SetDefault("jobs.shutdown_grace", "30s") + v.SetDefault("jobs.workspace_fetch.enabled", true) + v.SetDefault("jobs.workspace_fetch.interval", "5m") + v.SetDefault("jobs.worktree_prune.enabled", true) + v.SetDefault("jobs.worktree_prune.interval", "24h") + v.SetDefault("jobs.worktree_prune.idle_threshold", "720h") + v.SetDefault("jobs.worktree_prune.warning_lead", "72h") + v.SetDefault("jobs.attachment_cleanup.enabled", true) + v.SetDefault("jobs.attachment_cleanup.interval", "24h") + v.SetDefault("jobs.attachment_cleanup.retention", "720h") + v.SetDefault("jobs.job_reaper.enabled", true) + v.SetDefault("jobs.job_reaper.interval", "1m") + v.SetDefault("jobs.job_reaper.lease_timeout", "5m") + v.SetDefault("jobs.job_purge.enabled", true) + v.SetDefault("jobs.job_purge.interval", "24h") + v.SetDefault("jobs.job_purge.completed_retention", "168h") + v.SetDefault("notify.webhook.enabled", false) + v.SetDefault("notify.webhook.timeout", "10s") if configFile != "" { v.SetConfigFile(configFile) diff --git a/internal/user/domain.go b/internal/user/domain.go index 78bb5a4..5cdaf75 100644 --- a/internal/user/domain.go +++ b/internal/user/domain.go @@ -56,4 +56,6 @@ type Service interface { Get(ctx context.Context, id uuid.UUID) (*User, error) // Bootstrap 在系统首次启动时创建初始管理员账号，已存在同邮箱用户时为幂等。 Bootstrap(ctx context.Context, email, password string) (*User, error) + // ListAdmins 返回所有管理员用户，供内部模块（如 jobs runner）使用。 + ListAdmins(ctx context.Context) ([]*User, error) } diff --git a/internal/user/queries/users.sql b/internal/user/queries/users.sql index c802eb0..9c29974 100644 --- a/internal/user/queries/users.sql +++ b/internal/user/queries/users.sql @@ -36,3 +36,9 @@ DELETE FROM user_sessions WHERE token_hash = $1; -- name: DeleteExpiredSessions :exec DELETE FROM user_sessions WHERE expires_at <= now(); + +-- name: ListAdmins :many +SELECT id, email, password_hash, display_name, is_admin, created_at, updated_at + FROM users + WHERE is_admin = true + ORDER BY id; diff --git a/internal/user/repository.go b/internal/user/repository.go index 86a22bd..2cd5cb7 100644 --- a/internal/user/repository.go +++ b/internal/user/repository.go @@ -1,6 +1,6 @@ // Package user 内的 repository.go 是 sqlc 生成的查询包（usersqlc）之上的薄 -// 包装层。它的唯一职责是把 “驱动层类型”（pgtype.UUID / pgtype.Timestamptz / -// int32 等）和 “领域类型”（uuid.UUID / time.Time / int / *time.Time）做双向 +// 包装层。它的唯一职责是把 "驱动层类型"（pgtype.UUID / pgtype.Timestamptz / +// int32 等）和 "领域类型"（uuid.UUID / time.Time / int / *time.Time）做双向 // 转换，并把底层的 pgx.ErrNoRows 翻译为 errs.AppError，使 service 层可以面 // 向干净的领域接口编程，无需直接依赖驱动细节。 // @@ -35,6 +35,7 @@ type Repository interface { GetUserByEmail(ctx context.Context, email string) (*User, error) GetUserByID(ctx context.Context, id uuid.UUID) (*User, error) CountUsers(ctx context.Context) (int, error) + ListAdmins(ctx context.Context) ([]*User, error) CreateSession(ctx context.Context, s *Session) error GetSessionByTokenHash(ctx context.Context, hash []byte) (*Session, error) @@ -61,7 +62,7 @@ func toPgUUID(u uuid.UUID) pgtype.UUID { } // fromPgUUID 把 pgtype.UUID 转回领域 uuid.UUID。Valid=false 时返回 uuid.Nil， -// 让上层用零值判定 “无值”，无需关心驱动层的可空封装。 +// 让上层用零值判定 "无值"，无需关心驱动层的可空封装。 func fromPgUUID(p pgtype.UUID) uuid.UUID { if !p.Valid { return uuid.Nil @@ -168,8 +169,8 @@ func (r *pgRepo) CreateSession(ctx context.Context, s *Session) error { } // GetSessionByTokenHash 按 sha256 摘要反查会话。该方法主要给鉴权中间件 -// 使用，因此 ErrNoRows 翻译为 Unauthorized 而非 NotFound：从语义看 “找不到 -// 会话” 等价于 “未认证”，不应该把内部存在性细节透给客户端。 +// 使用，因此 ErrNoRows 翻译为 Unauthorized 而非 NotFound：从语义看 "找不到 +// 会话" 等价于 "未认证"，不应该把内部存在性细节透给客户端。 func (r *pgRepo) GetSessionByTokenHash(ctx context.Context, hash []byte) (*Session, error) { row, err := r.q.GetSessionByTokenHash(ctx, hash) if err != nil { @@ -191,10 +192,23 @@ func (r *pgRepo) TouchSession(ctx context.Context, hash []byte) error { } // DeleteSessionByTokenHash 删除指定哈希的会话。删除不存在的行不算错误， -// 调用方可以放心当作 “登出幂等” 使用。 +// 调用方可以放心当作 "登出幂等" 使用。 func (r *pgRepo) DeleteSessionByTokenHash(ctx context.Context, hash []byte) error { if err := r.q.DeleteSessionByTokenHash(ctx, hash); err != nil { return errs.Wrap(err, errs.CodeInternal, "delete session by token hash") } return nil } + +// ListAdmins 返回所有 is_admin=true 的用户，供 jobs runner 等内部模块使用。 +func (r *pgRepo) ListAdmins(ctx context.Context) ([]*User, error) { + rows, err := r.q.ListAdmins(ctx) + if err != nil { + return nil, errs.Wrap(err, errs.CodeInternal, "list admins") + } + out := make([]*User, 0, len(rows)) + for _, row := range rows { + out = append(out, rowToUser(row)) + } + return out, nil +} diff --git a/internal/user/service.go b/internal/user/service.go index 8172561..2a8661d 100644 --- a/internal/user/service.go +++ b/internal/user/service.go @@ -172,3 +172,8 @@ func (s *service) Bootstrap(ctx context.Context, email, password string) (*User, } return s.repo.CreateUser(ctx, u) } + +// ListAdmins 返回所有管理员用户，供内部模块（如 jobs runner）使用。 +func (s *service) ListAdmins(ctx context.Context) ([]*User, error) { + return s.repo.ListAdmins(ctx) +} diff --git a/internal/user/service_test.go b/internal/user/service_test.go index ed9fb4d..adfa720 100644 --- a/internal/user/service_test.go +++ b/internal/user/service_test.go @@ -48,6 +48,7 @@ func (r *fakeRepo) GetUserByID(_ context.Context, id uuid.UUID) (*User, error) { } func (r *fakeRepo) CountUsers(_ context.Context) (int, error) { return len(r.users), nil } +func (r *fakeRepo) ListAdmins(_ context.Context) ([]*User, error) { return nil, nil } func (r *fakeRepo) CreateSession(_ context.Context, s *Session) error { r.sessions[string(s.TokenHash)] = s diff --git a/internal/user/sqlc/users.sql.go b/internal/user/sqlc/users.sql.go index c26a84e..ca13df5 100644 --- a/internal/user/sqlc/users.sql.go +++ b/internal/user/sqlc/users.sql.go @@ -160,6 +160,41 @@ func (q *Queries) GetUserByID(ctx context.Context, id pgtype.UUID) (User, error) return i, err } +const listAdmins = `-- name: ListAdmins :many +SELECT id, email, password_hash, display_name, is_admin, created_at, updated_at + FROM users + WHERE is_admin = true + ORDER BY id +` + +func (q *Queries) ListAdmins(ctx context.Context) ([]User, error) { + rows, err := q.db.Query(ctx, listAdmins) + if err != nil { + return nil, err + } + defer rows.Close() + var items []User + for rows.Next() { + var i User + if err := rows.Scan( + &i.ID, + &i.Email, + &i.PasswordHash, + &i.DisplayName, + &i.IsAdmin, + &i.CreatedAt, + &i.UpdatedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const touchSession = `-- name: TouchSession :exec UPDATE user_sessions SET last_seen_at = now() diff --git a/internal/workspace/domain.go b/internal/workspace/domain.go index b7685f7..ede1fe8 100644 --- a/internal/workspace/domain.go +++ b/internal/workspace/domain.go @@ -102,6 +102,15 @@ type Credential struct { UpdatedAt time.Time } +// FetchTarget 是 workspace 包暴露给 jobs 模块的"周期 fetch 目标"投影。 +// owner_id 通过 JOIN projects 取，避免 workspaces 表加冗余列。 +type FetchTarget struct { + ID uuid.UUID + OwnerID uuid.UUID + Name string + MainPath string +} + // Caller 表示发起请求的用户上下文。Service 据此判定鉴权与 holder 字符串。 type Caller struct { UserID uuid.UUID diff --git a/internal/workspace/queries/workspaces.sql b/internal/workspace/queries/workspaces.sql index 5eefc6d..f06a329 100644 --- a/internal/workspace/queries/workspaces.sql +++ b/internal/workspace/queries/workspaces.sql @@ -41,3 +41,23 @@ SET sync_status = 'error', last_sync_error = 'process_restarted_during_' || sync_status, updated_at = now() WHERE sync_status IN ('cloning', 'syncing'); + +-- name: ListFetchTargets :many +SELECT w.id, p.owner_id, w.name, w.main_path + FROM workspaces w + JOIN projects p ON p.id = w.project_id + WHERE w.archived_at IS NULL + AND w.sync_status IN ('idle','error') + ORDER BY w.id; + +-- name: MarkSyncingCAS :execrows +UPDATE workspaces SET sync_status='syncing', updated_at=now() + WHERE id=$1 AND sync_status IN ('idle','error'); + +-- name: MarkFetchResult :exec +UPDATE workspaces + SET sync_status = CASE WHEN $2::bool THEN 'idle' ELSE 'error' END, + last_synced_at = CASE WHEN $2::bool THEN now() ELSE last_synced_at END, + last_sync_error = $3, + updated_at = now() + WHERE id = $1; diff --git a/internal/workspace/repository.go b/internal/workspace/repository.go index a023687..2936cfc 100644 --- a/internal/workspace/repository.go +++ b/internal/workspace/repository.go @@ -57,6 +57,11 @@ type Repository interface { FinishPruneSuccess(ctx context.Context, id uuid.UUID) error FinishPruneRollback(ctx context.Context, id uuid.UUID) error + // Workspace fetch (jobs runner) + ListFetchTargets(ctx context.Context) ([]FetchTarget, error) + MarkSyncing(ctx context.Context, id uuid.UUID) (bool, error) + MarkFetchResult(ctx context.Context, id uuid.UUID, success bool, errMsg string) error + // Credential UpsertCredential(ctx context.Context, c *Credential) (*Credential, error) GetCredentialByWorkspace(ctx context.Context, wsID uuid.UUID) (*Credential, error) @@ -242,6 +247,49 @@ func (r *pgRepo) ResetStuckSyncStatuses(ctx context.Context) error { return nil } +// ===== Workspace fetch (jobs runner) ===== + +func (r *pgRepo) ListFetchTargets(ctx context.Context) ([]FetchTarget, error) { + rows, err := r.q.ListFetchTargets(ctx) + if err != nil { + return nil, errs.Wrap(err, errs.CodeInternal, "list fetch targets") + } + out := make([]FetchTarget, 0, len(rows)) + for _, row := range rows { + out = append(out, FetchTarget{ + ID: fromPgUUID(row.ID), + OwnerID: fromPgUUID(row.OwnerID), + Name: row.Name, + MainPath: row.MainPath, + }) + } + return out, nil +} + +func (r *pgRepo) MarkSyncing(ctx context.Context, id uuid.UUID) (bool, error) { + n, err := r.q.MarkSyncingCAS(ctx, toPgUUID(id)) + if err != nil { + return false, errs.Wrap(err, errs.CodeInternal, "mark syncing") + } + return n > 0, nil +} + +func (r *pgRepo) MarkFetchResult(ctx context.Context, id uuid.UUID, success bool, errMsg string) error { + var lastErr *string + if errMsg != "" { + lastErr = &errMsg + } + err := r.q.MarkFetchResult(ctx, workspacesqlc.MarkFetchResultParams{ + ID: toPgUUID(id), + Column2: success, + LastSyncError: lastErr, + }) + if err != nil { + return errs.Wrap(err, errs.CodeInternal, "mark fetch result") + } + return nil +} + // ===== Worktree ===== func rowToWorktree(r workspacesqlc.WorkspaceWorktree) *Worktree { diff --git a/internal/workspace/sqlc/workspaces.sql.go b/internal/workspace/sqlc/workspaces.sql.go index 80e8286..2e1cdca 100644 --- a/internal/workspace/sqlc/workspaces.sql.go +++ b/internal/workspace/sqlc/workspaces.sql.go @@ -128,6 +128,47 @@ func (q *Queries) GetWorkspaceBySlug(ctx context.Context, arg GetWorkspaceBySlug return i, err } +const listFetchTargets = `-- name: ListFetchTargets :many +SELECT w.id, p.owner_id, w.name, w.main_path + FROM workspaces w + JOIN projects p ON p.id = w.project_id + WHERE w.archived_at IS NULL + AND w.sync_status IN ('idle','error') + ORDER BY w.id +` + +type ListFetchTargetsRow struct { + ID pgtype.UUID `json:"id"` + OwnerID pgtype.UUID `json:"owner_id"` + Name string `json:"name"` + MainPath string `json:"main_path"` +} + +func (q *Queries) ListFetchTargets(ctx context.Context) ([]ListFetchTargetsRow, error) { + rows, err := q.db.Query(ctx, listFetchTargets) + if err != nil { + return nil, err + } + defer rows.Close() + var items []ListFetchTargetsRow + for rows.Next() { + var i ListFetchTargetsRow + if err := rows.Scan( + &i.ID, + &i.OwnerID, + &i.Name, + &i.MainPath, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const listWorkspacesByProject = `-- name: ListWorkspacesByProject :many SELECT id, project_id, slug, name, description, git_remote_url, default_branch, main_path, sync_status, last_synced_at, last_sync_error, created_at, updated_at FROM workspaces WHERE project_id = $1 ORDER BY created_at DESC @@ -167,6 +208,39 @@ func (q *Queries) ListWorkspacesByProject(ctx context.Context, projectID pgtype. return items, nil } +const markFetchResult = `-- name: MarkFetchResult :exec +UPDATE workspaces + SET sync_status = CASE WHEN $2::bool THEN 'idle' ELSE 'error' END, + last_synced_at = CASE WHEN $2::bool THEN now() ELSE last_synced_at END, + last_sync_error = $3, + updated_at = now() + WHERE id = $1 +` + +type MarkFetchResultParams struct { + ID pgtype.UUID `json:"id"` + Column2 bool `json:"column_2"` + LastSyncError *string `json:"last_sync_error"` +} + +func (q *Queries) MarkFetchResult(ctx context.Context, arg MarkFetchResultParams) error { + _, err := q.db.Exec(ctx, markFetchResult, arg.ID, arg.Column2, arg.LastSyncError) + return err +} + +const markSyncingCAS = `-- name: MarkSyncingCAS :execrows +UPDATE workspaces SET sync_status='syncing', updated_at=now() + WHERE id=$1 AND sync_status IN ('idle','error') +` + +func (q *Queries) MarkSyncingCAS(ctx context.Context, id pgtype.UUID) (int64, error) { + result, err := q.db.Exec(ctx, markSyncingCAS, id) + if err != nil { + return 0, err + } + return result.RowsAffected(), nil +} + const resetStuckSyncStatuses = `-- name: ResetStuckSyncStatuses :exec UPDATE workspaces SET sync_status = 'error', diff --git a/internal/workspace/workspace_service_test.go b/internal/workspace/workspace_service_test.go index ae5502a..5b4c46c 100644 --- a/internal/workspace/workspace_service_test.go +++ b/internal/workspace/workspace_service_test.go @@ -76,6 +76,11 @@ func (f *fakeRepo) DeleteWorkspace(_ context.Context, id uuid.UUID) error { return nil } func (f *fakeRepo) ResetStuckSyncStatuses(_ context.Context) error { return nil } +func (f *fakeRepo) ListFetchTargets(_ context.Context) ([]FetchTarget, error) { return nil, nil } +func (f *fakeRepo) MarkSyncing(_ context.Context, _ uuid.UUID) (bool, error) { return false, nil } +func (f *fakeRepo) MarkFetchResult(_ context.Context, _ uuid.UUID, _ bool, _ string) error { + return nil +} func (f *fakeRepo) GetWorktreeByID(_ context.Context, id uuid.UUID) (*Worktree, error) { if w, ok := f.wts[id]; ok { return w, nil