From 2174667d9fdc84a3a03506808d44e265f19f857c Mon Sep 17 00:00:00 2001 From: ranxi2001 Date: Thu, 22 Oct 2026 02:06:26 +0800 Subject: [PATCH 1/2] feat(accounts): add persistent one-shot schedulable timers --- backend/internal/accounttimer/store.go | 469 ++++++++++++++++++ .../accounttimer/store_integration_test.go | 447 +++++++++++++++++ backend/internal/accounttimer/store_test.go | 264 ++++++++++ backend/internal/accounttimer/time.go | 157 ++++++ backend/internal/accounttimer/time_test.go | 74 +++ .../internal/handler/admin/account_handler.go | 6 +- .../admin/account_schedulable_timer.go | 41 ++ .../handler/admin/accounttimerapi/handler.go | 121 +++++ .../admin/accounttimerapi/handler_test.go | 129 +++++ backend/internal/repository/account_repo.go | 49 +- .../repository/account_schedulable_timer.go | 25 + backend/internal/server/routes/admin.go | 4 + .../service/account_expiry_service.go | 27 + .../service/account_expiry_timer_test.go | 78 +++ .../service/account_schedulable_timer.go | 14 + backend/internal/service/admin_account.go | 8 + .../295_account_schedulable_timers.sql | 35 ++ docs/account-schedulable-timers.md | 58 +++ frontend/src/api/admin/schedulableTimers.ts | 66 +++ .../account/SchedulableTimerControl.vue | 202 ++++++++ .../__tests__/SchedulableTimerControl.spec.ts | 109 ++++ .../src/i18n/locales/en/admin/accounts.ts | 61 +++ .../src/i18n/locales/zh/admin/accounts.ts | 61 +++ frontend/src/views/admin/AccountsView.vue | 16 + .../AccountsView.schedulableTimer.spec.ts | 186 +++++++ 25 files changed, 2703 insertions(+), 4 deletions(-) create mode 100644 backend/internal/accounttimer/store.go create mode 100644 backend/internal/accounttimer/store_integration_test.go create mode 100644 backend/internal/accounttimer/store_test.go create mode 100644 backend/internal/accounttimer/time.go create mode 100644 backend/internal/accounttimer/time_test.go create mode 100644 backend/internal/handler/admin/account_schedulable_timer.go create mode 100644 backend/internal/handler/admin/accounttimerapi/handler.go create mode 100644 backend/internal/handler/admin/accounttimerapi/handler_test.go create mode 100644 backend/internal/repository/account_schedulable_timer.go create mode 100644 backend/internal/service/account_expiry_timer_test.go create mode 100644 backend/internal/service/account_schedulable_timer.go create mode 100644 backend/migrations/295_account_schedulable_timers.sql create mode 100644 docs/account-schedulable-timers.md create mode 100644 frontend/src/api/admin/schedulableTimers.ts create mode 100644 frontend/src/components/account/SchedulableTimerControl.vue create mode 100644 frontend/src/components/account/__tests__/SchedulableTimerControl.spec.ts create mode 100644 frontend/src/views/admin/__tests__/AccountsView.schedulableTimer.spec.ts diff --git a/backend/internal/accounttimer/store.go b/backend/internal/accounttimer/store.go new file mode 100644 index 000000000000..00c6791e8ff7 --- /dev/null +++ b/backend/internal/accounttimer/store.go @@ -0,0 +1,469 @@ +package accounttimer + +import ( + "context" + "database/sql" + "errors" + "fmt" + "time" +) + +type Task struct { + AccountID int64 `json:"account_id"` + Action string `json:"action"` + Version int64 `json:"version"` + ExecuteAt time.Time `json:"execute_at"` + Timezone string `json:"timezone"` + LocalDate string `json:"local_date"` + LocalTime string `json:"local_time"` + Offset string `json:"offset"` + RequestedBy int64 `json:"requested_by"` + Status string `json:"status"` + Attempts int `json:"attempts"` + ActualAt *time.Time `json:"actual_at"` + Reason string `json:"reason"` +} +type Audit struct { + Action string `json:"action"` + Version int64 `json:"version"` + ActorID int64 `json:"actor_id"` + ExecuteAt time.Time `json:"execute_at"` + Timezone string `json:"timezone"` + ActualAt time.Time `json:"actual_at"` + Result string `json:"result"` + Reason string `json:"reason"` +} +type Snapshot struct { + Tasks []Task `json:"tasks"` + History []Audit `json:"history"` + ServerNow time.Time `json:"server_now"` + PollSeconds int `json:"poll_seconds"` + OpenGraceSeconds int `json:"open_grace_seconds"` + MaxDelayMinutes int64 `json:"max_delay_minutes"` +} +type SaveRequest struct { + Input + Action string `json:"action"` + ExpectedVersion int64 `json:"expected_version"` + // Required for fixed times: prove that the caller confirmed the server preview. + ConfirmedAt *time.Time `json:"confirmed_at"` +} +type Repository interface { + Preview(Input) (Preview, error) + List(context.Context, int64) (Snapshot, error) + Save(context.Context, int64, int64, SaveRequest) (Task, error) + Cancel(context.Context, int64, int64, string, int64) error + RunDue(context.Context) error +} +type Store struct { + DB *sql.DB + Now func() time.Time + Policy Policy + Enqueue func(context.Context, *sql.Tx, int64) error + AfterCommit func(context.Context, int64) +} + +func (s *Store) now() time.Time { + if s.Now != nil { + return s.Now().UTC() + } + return time.Now().UTC() +} +func (s *Store) policy() Policy { + if s.Policy.MaxDelay <= 0 { + return DefaultPolicy() + } + return s.Policy +} +func (s *Store) Preview(in Input) (Preview, error) { return Resolve(in, s.now(), s.policy()) } + +const taskColumns = "account_id, action, version, execute_at, timezone, requested_by, status, attempts, actual_at, reason" + +func fillLocalTime(t *Task) { + if loc, err := time.LoadLocation(t.Timezone); err == nil { + local := t.ExecuteAt.In(loc) + t.LocalDate = local.Format("2006-01-02") + t.LocalTime = local.Format("15:04") + t.Offset = local.Format("-07:00") + } +} +func actionValid(a string) bool { return a == "open" || a == "close" } + +type scanner interface{ Scan(...any) error } + +func scanTask(row scanner) (Task, error) { + var t Task + err := row.Scan(&t.AccountID, &t.Action, &t.Version, &t.ExecuteAt, &t.Timezone, &t.RequestedBy, &t.Status, &t.Attempts, &t.ActualAt, &t.Reason) + if err == nil { + t.ExecuteAt = t.ExecuteAt.UTC() + fillLocalTime(&t) + if t.ActualAt != nil { + utc := t.ActualAt.UTC() + t.ActualAt = &utc + } + } + return t, err +} + +type accountState struct { + status string + expires *time.Time + deleted *time.Time +} + +func lockAccount(ctx context.Context, tx *sql.Tx, id int64, skip bool) (accountState, error) { + var a accountState + suffix := " FOR UPDATE" + if skip { + suffix += " SKIP LOCKED" + } + err := tx.QueryRowContext(ctx, "SELECT status, expires_at, deleted_at FROM accounts WHERE id = $1"+suffix, id).Scan(&a.status, &a.expires, &a.deleted) + return a, err +} +func audit(ctx context.Context, tx *sql.Tx, t Task, actor int64, now time.Time, result, reason string) error { + _, err := tx.ExecContext(ctx, `INSERT INTO account_schedulable_timer_audit + (account_id,action,version,actor_id,execute_at,timezone,actual_at,result,reason) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9)`, t.AccountID, t.Action, t.Version, actor, t.ExecuteAt, t.Timezone, now, result, reason) + return err +} +func (s *Store) Save(ctx context.Context, id, actor int64, req SaveRequest) (Task, error) { + var zero Task + if id <= 0 || actor <= 0 || !actionValid(req.Action) || req.ExpectedVersion < 0 { + return zero, invalid("request") + } + tx, err := s.DB.BeginTx(ctx, nil) + if err != nil { + return zero, err + } + defer func() { _ = tx.Rollback() }() + a, err := lockAccount(ctx, tx, id, false) + if errors.Is(err, sql.ErrNoRows) || (err == nil && a.deleted != nil) { + return zero, invalid("account_deleted") + } + if err != nil { + return zero, err + } + now := s.now() + preview, err := Resolve(req.Input, now, s.policy()) + if err != nil { + return zero, err + } + if req.Mode == "absolute" && (req.Date == "" || req.ConfirmedAt == nil || !req.ConfirmedAt.Equal(preview.ExecuteAt)) { + return zero, invalid("confirmation") + } + var version int64 + err = tx.QueryRowContext(ctx, "SELECT version FROM account_schedulable_timers WHERE account_id=$1 AND action=$2 FOR UPDATE", id, req.Action).Scan(&version) + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return zero, err + } + if version != req.ExpectedVersion { + return zero, invalid("version") + } + var conflict bool + err = tx.QueryRowContext(ctx, `SELECT EXISTS(SELECT 1 FROM account_schedulable_timers WHERE account_id=$1 AND action<>$2 AND status='pending' AND execute_at=$3)`, id, req.Action, preview.ExecuteAt).Scan(&conflict) + if err != nil { + return zero, err + } + if conflict { + return zero, invalid("conflict") + } + t := Task{AccountID: id, Action: req.Action, Version: version + 1, ExecuteAt: preview.ExecuteAt, Timezone: req.Timezone, RequestedBy: actor, Status: "pending"} + _, err = tx.ExecContext(ctx, `INSERT INTO account_schedulable_timers + (account_id,action,version,execute_at,timezone,requested_by,status,attempts,next_attempt_at,actual_at,reason) + VALUES ($1,$2,$3,$4,$5,$6,'pending',0,$4,NULL,'') + ON CONFLICT (account_id,action) DO UPDATE SET version=EXCLUDED.version,execute_at=EXCLUDED.execute_at, + timezone=EXCLUDED.timezone,requested_by=EXCLUDED.requested_by,status='pending',attempts=0,next_attempt_at=EXCLUDED.execute_at,actual_at=NULL,reason=''`, id, t.Action, t.Version, t.ExecuteAt, t.Timezone, actor) + if err != nil { + return zero, err + } + fillLocalTime(&t) + result := "scheduled" + if version > 0 { + result = "replaced" + } + if err = audit(ctx, tx, t, actor, now, result, ""); err != nil { + return zero, err + } + if err = tx.Commit(); err != nil { + return zero, err + } + return t, nil +} +func (s *Store) Cancel(ctx context.Context, id, actor int64, action string, version int64) error { + if id <= 0 || actor <= 0 || !actionValid(action) || version <= 0 { + return invalid("request") + } + tx, err := s.DB.BeginTx(ctx, nil) + if err != nil { + return err + } + defer func() { _ = tx.Rollback() }() + _, err = lockAccount(ctx, tx, id, false) + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return err + } + t, err := scanTask(tx.QueryRowContext(ctx, "SELECT "+taskColumns+" FROM account_schedulable_timers WHERE account_id=$1 AND action=$2 FOR UPDATE", id, action)) + if errors.Is(err, sql.ErrNoRows) { + return invalid("version") + } + if err != nil { + return err + } + if t.Version != version || t.Status != "pending" { + return invalid("version") + } + now := s.now() + _, err = tx.ExecContext(ctx, `UPDATE account_schedulable_timers SET status='cancelled',version=version+1,actual_at=$3,reason='admin_cancelled' WHERE account_id=$1 AND action=$2`, id, action, now) + if err != nil { + return err + } + if err = audit(ctx, tx, t, actor, now, "cancelled", "admin_cancelled"); err != nil { + return err + } + return tx.Commit() +} +func (s *Store) List(ctx context.Context, id int64) (Snapshot, error) { + p := s.policy() + out := Snapshot{Tasks: []Task{}, History: []Audit{}, ServerNow: s.now(), PollSeconds: int(PollInterval / time.Second), OpenGraceSeconds: int(p.OpenGrace / time.Second), MaxDelayMinutes: int64(p.MaxDelay / time.Minute)} + rows, err := s.DB.QueryContext(ctx, "SELECT "+taskColumns+" FROM account_schedulable_timers WHERE account_id=$1 ORDER BY action", id) + if err != nil { + return out, err + } + for rows.Next() { + t, e := scanTask(rows) + if e != nil { + _ = rows.Close() + return out, e + } + out.Tasks = append(out.Tasks, t) + } + err = rows.Err() + _ = rows.Close() + if err != nil { + return out, err + } + rows, err = s.DB.QueryContext(ctx, `SELECT action,version,actor_id,execute_at,timezone,actual_at,result,reason FROM account_schedulable_timer_audit WHERE account_id=$1 ORDER BY id DESC LIMIT 20`, id) + if err != nil { + return out, err + } + defer func() { _ = rows.Close() }() + for rows.Next() { + var a Audit + if err = rows.Scan(&a.Action, &a.Version, &a.ActorID, &a.ExecuteAt, &a.Timezone, &a.ActualAt, &a.Result, &a.Reason); err != nil { + return out, err + } + a.ExecuteAt = a.ExecuteAt.UTC() + a.ActualAt = a.ActualAt.UTC() + out.History = append(out.History, a) + } + return out, rows.Err() +} +func (s *Store) RunDue(ctx context.Context) error { + rows, err := s.DB.QueryContext(ctx, "SELECT "+taskColumns+" FROM account_schedulable_timers WHERE status='pending' AND next_attempt_at <= $1 ORDER BY next_attempt_at,account_id,action LIMIT 50", s.now()) + if err != nil { + return err + } + tasks := []Task{} + for rows.Next() { + t, e := scanTask(rows) + if e != nil { + _ = rows.Close() + return e + } + tasks = append(tasks, t) + } + err = rows.Err() + _ = rows.Close() + if err != nil { + return err + } + var errs []error + for _, t := range tasks { + if e := s.execute(ctx, t); e != nil { + errs = append(errs, e) + } + if ctx.Err() != nil { + break + } + } + return errors.Join(errs...) +} +func (s *Store) execute(ctx context.Context, candidate Task) error { + tx, err := s.DB.BeginTx(ctx, nil) + if err != nil { + return err + } + defer func() { _ = tx.Rollback() }() + a, err := lockAccount(ctx, tx, candidate.AccountID, true) + if errors.Is(err, sql.ErrNoRows) { + // SKIP LOCKED must not mistake a busy account for a deleted one. + var exists bool + if e := tx.QueryRowContext(ctx, "SELECT EXISTS(SELECT 1 FROM accounts WHERE id=$1)", candidate.AccountID).Scan(&exists); e != nil { + return e + } + if exists { + return nil + } + deleted := s.now() + a.deleted = &deleted + } else if err != nil { + return err + } + now := s.now() + t, err := scanTask(tx.QueryRowContext(ctx, "SELECT "+taskColumns+` FROM account_schedulable_timers WHERE account_id=$1 AND action=$2 AND next_attempt_at <= $3 AND NOT EXISTS ( + SELECT 1 FROM account_schedulable_timers earlier WHERE earlier.account_id=account_schedulable_timers.account_id + AND earlier.status='pending' AND earlier.execute_at < account_schedulable_timers.execute_at) FOR UPDATE`, candidate.AccountID, candidate.Action, now)) + if errors.Is(err, sql.ErrNoRows) { + return nil + } + if err != nil { + return err + } + if t.Version != candidate.Version || t.Status != "pending" || t.ExecuteAt.After(now) { + return nil + } + reason := Decision(t.Action == "open", t.ExecuteAt, now, a.status, a.expires, a.deleted != nil, s.policy()) + result := "executed" + if reason != "" { + result = "skipped" + } else { + // Do not touch status, expiry, credentials, quotas, cooldown or quality fields. + _, err = tx.ExecContext(ctx, "UPDATE accounts SET schedulable=$1, updated_at=$2 WHERE id=$3 AND deleted_at IS NULL", t.Action == "open", now, t.AccountID) + if err == nil && s.Enqueue != nil { + err = s.Enqueue(ctx, tx, t.AccountID) + } + if err != nil { + _ = tx.Rollback() + return s.retry(ctx, t) + } + } + _, err = tx.ExecContext(ctx, `UPDATE account_schedulable_timers SET status=$3,actual_at=$4,reason=$5 WHERE account_id=$1 AND action=$2 AND version=$6`, t.AccountID, t.Action, result, now, reason, t.Version) + if err == nil { + err = audit(ctx, tx, t, t.RequestedBy, now, result, reason) + } + if err != nil { + _ = tx.Rollback() + return s.retry(ctx, t) + } + if err = tx.Commit(); err != nil { + return err + } // Ambiguous commit: reread pending state next poll, never replay blindly. + if result == "executed" && s.AfterCommit != nil { + s.AfterCommit(ctx, t.AccountID) + } + return nil +} +func (s *Store) retry(ctx context.Context, t Task) error { + tx, err := s.DB.BeginTx(ctx, nil) + if err != nil { + return err + } + defer func() { _ = tx.Rollback() }() + _, err = lockAccount(ctx, tx, t.AccountID, false) + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return err + } + now := s.now() + attempts := t.Attempts + 1 + status := "pending" + if attempts >= 3 { + status = "failed" + } + result, err := tx.ExecContext(ctx, `UPDATE account_schedulable_timers SET attempts=$4,status=$5,next_attempt_at=$6,actual_at=$7,reason='execution_error' + WHERE account_id=$1 AND action=$2 AND version=$3 AND status='pending' AND attempts=$8`, t.AccountID, t.Action, t.Version, attempts, status, now.Add(time.Duration(1<0), execute_at timestamptz NOT NULL, timezone text NOT NULL, + requested_by bigint NOT NULL, status text NOT NULL CHECK(status IN ('pending','executed','skipped','cancelled','failed')), + attempts integer NOT NULL DEFAULT 0 CHECK(attempts BETWEEN 0 AND 3), next_attempt_at timestamptz NOT NULL, + actual_at timestamptz, reason text NOT NULL DEFAULT '', PRIMARY KEY(account_id,action) +); +CREATE INDEX timer_fixture_due ON account_schedulable_timers(next_attempt_at,account_id) WHERE status='pending'; +CREATE UNIQUE INDEX timer_fixture_conflict ON account_schedulable_timers(account_id,execute_at) WHERE status='pending'; +CREATE TABLE account_schedulable_timer_audit ( + id bigserial PRIMARY KEY, account_id bigint NOT NULL, action text NOT NULL, version bigint NOT NULL, + actor_id bigint NOT NULL, execute_at timestamptz NOT NULL, timezone text NOT NULL, + actual_at timestamptz NOT NULL, result text NOT NULL, reason text NOT NULL DEFAULT '' +); +CREATE TABLE fixture_outbox(id bigserial PRIMARY KEY, account_id bigint NOT NULL); +INSERT INTO accounts(id,temp_unschedulable_until,rate_limit_reset_at) VALUES(7,'2030-01-01Z','2030-01-02Z'); +` + +func integrationStore(t *testing.T) (*Store, *time.Time, *atomic.Int64) { + t.Helper() + dsn := os.Getenv("ACCOUNT_TIMER_TEST_DSN") + if dsn == "" { + t.Skip("ACCOUNT_TIMER_TEST_DSN not set; disposable loopback fixture required") + } + u, err := url.Parse(dsn) + if err != nil || (u.Hostname() != "127.0.0.1" && u.Hostname() != "localhost") || u.Path != "/schedulable_timer_fixture" { + t.Fatal("test DSN must target loopback /schedulable_timer_fixture") + } + root, err := sql.Open("postgres", dsn) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = root.Close() }) + schema := fmt.Sprintf("timer_fixture_%d", time.Now().UnixNano()) + if _, err = root.Exec("CREATE SCHEMA " + pq.QuoteIdentifier(schema)); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + _, e := root.Exec("DROP SCHEMA " + pq.QuoteIdentifier(schema) + " CASCADE") + if e != nil { + t.Error(e) + } + }) + query := u.Query() + query.Set("search_path", schema) + query.Set("TimeZone", "Asia/Shanghai") + u.RawQuery = query.Encode() + db, err := sql.Open("postgres", u.String()) + if err != nil { + t.Fatal(err) + } + db.SetMaxOpenConns(6) + t.Cleanup(func() { _ = db.Close() }) + if _, err = db.Exec(timerFixture); err != nil { + t.Fatal(err) + } + now := time.Date(2026, 10, 22, 0, 0, 0, 0, time.UTC) + notified := &atomic.Int64{} + s := &Store{DB: db, Now: func() time.Time { return now }, Policy: DefaultPolicy(), Enqueue: func(ctx context.Context, tx *sql.Tx, id int64) error { + _, err := tx.ExecContext(ctx, "INSERT INTO fixture_outbox(account_id) VALUES($1)", id) + return err + }, AfterCommit: func(context.Context, int64) { notified.Add(1) }} + return s, &now, notified +} +func saveRelative(t *testing.T, s *Store, action string, hours int64) Task { + t.Helper() + task, err := s.Save(context.Background(), 7, 9, SaveRequest{Input: Input{Mode: "relative", Hours: hours, Timezone: "Asia/Shanghai"}, Action: action}) + if err != nil { + t.Fatal(err) + } + return task +} +func snapshotTask(t *testing.T, s *Store, action string) Task { + t.Helper() + snap, err := s.List(context.Background(), 7) + if err != nil { + t.Fatal(err) + } + for _, task := range snap.Tasks { + if task.Action == action { + return task + } + } + t.Fatal("missing task") + return Task{} +} +func scalar(t *testing.T, db *sql.DB, query string) int { + t.Helper() + var n int + if err := db.QueryRow(query).Scan(&n); err != nil { + t.Fatal(err) + } + return n +} +func TestIntegrationPersistRestartMultiInstanceAndSafetyGates(t *testing.T) { + s, now, notified := integrationStore(t) + open := saveRelative(t, s, "open", 8) + saveRelative(t, s, "close", 9) + if open.ExecuteAt.Format(time.RFC3339) != "2026-10-22T08:00:00Z" || open.Timezone != "Asia/Shanghai" { + t.Fatal(open) + } + listed := snapshotTask(t, s, "open") + if listed.ExecuteAt.Location() != time.UTC || listed.LocalDate != "2026-10-22" || listed.LocalTime != "16:00" || listed.Offset != "+08:00" { + t.Fatalf("server/database timezone leaked into UTC output: %+v", listed) + } + // A fresh Store is a simulated process restart; the DB owns pending work. + restarted := *s + *now = now.Add(8 * time.Hour) + var wg sync.WaitGroup + errs := make(chan error, 8) + for i := 0; i < 8; i++ { + wg.Add(1) + go func() { defer wg.Done(); errs <- restarted.RunDue(context.Background()) }() + } + wg.Wait() + close(errs) + for err := range errs { + if err != nil { + t.Fatal(err) + } + } + if scalar(t, s.DB, "SELECT count(*) FROM accounts WHERE schedulable") != 1 || scalar(t, s.DB, "SELECT count(*) FROM fixture_outbox") != 1 || notified.Load() != 1 { + t.Fatal("logical effect and cache publication must happen once") + } + if scalar(t, s.DB, "SELECT count(*) FROM account_schedulable_timer_audit WHERE result='executed' AND actor_id=9") != 1 { + t.Fatal("execution audit missing/duplicated") + } + if scalar(t, s.DB, `SELECT count(*) FROM accounts WHERE quota_used=17 AND quality_reason='quality-isolation' AND manual_ip_acl='deny-fixture' AND balance=23 AND temp_unschedulable_until='2030-01-01Z' AND rate_limit_reset_at='2030-01-02Z'`) != 1 { + t.Fatal("safety or financial sentinel changed") + } + *now = now.Add(24 * time.Hour) + if err := s.RunDue(context.Background()); err != nil { + t.Fatal(err) + } + if scalar(t, s.DB, "SELECT count(*) FROM accounts WHERE schedulable") != 0 || snapshotTask(t, s, "close").Status != "executed" { + t.Fatal("offline close must catch up") + } +} +func TestIntegrationConflictsReplacementCancellationAndManualOverride(t *testing.T) { + s, now, _ := integrationStore(t) + ctx := context.Background() + open := saveRelative(t, s, "open", 8) + _, err := s.Save(ctx, 7, 9, SaveRequest{Input: Input{Mode: "relative", Hours: 8, Timezone: "UTC"}, Action: "close"}) + if ErrorCode(err) != "conflict" { + t.Fatalf("%v", err) + } + closeTask := saveRelative(t, s, "close", 9) + edit := SaveRequest{Input: Input{Mode: "relative", Hours: 10, Timezone: "UTC"}, Action: "open", ExpectedVersion: open.Version} + replacement, err := s.Save(ctx, 7, 10, edit) + if err != nil { + t.Fatal(err) + } + if _, err = s.Save(ctx, 7, 9, edit); ErrorCode(err) != "version" { + t.Fatalf("old editor accepted: %v", err) + } + if err = s.Cancel(ctx, 7, 9, "open", open.Version); ErrorCode(err) != "version" { + t.Fatalf("old cancel accepted: %v", err) + } + *now = now.Add(8 * time.Hour) + if err = s.execute(ctx, open); err != nil { + t.Fatal(err) + } + if scalar(t, s.DB, "SELECT count(*) FROM fixture_outbox") != 0 { + t.Fatal("stale claimed task changed account") + } + if err = s.Cancel(ctx, 7, 9, "close", closeTask.Version); err != nil { + t.Fatal(err) + } + if snapshotTask(t, s, "open").Version != replacement.Version { + t.Fatal("opposite cancellation affected open") + } + if err = s.Manual(ctx, 7, 9, false); err != nil { + t.Fatal(err) + } + *now = now.Add(24 * time.Hour) + if err = s.RunDue(ctx); err != nil { + t.Fatal(err) + } + if scalar(t, s.DB, "SELECT count(*) FROM account_schedulable_timers WHERE status='pending'") != 0 || scalar(t, s.DB, "SELECT count(*) FROM accounts WHERE schedulable") != 0 { + t.Fatal("old timers reopened a manual close") + } +} +func TestIntegrationSkipDisabledExpiredDeletedAndLateOpens(t *testing.T) { + for _, kind := range []string{"disabled", "expired", "soft_deleted", "hard_deleted", "late"} { + t.Run(kind, func(t *testing.T) { + s, now, _ := integrationStore(t) + saveRelative(t, s, "open", 1) + *now = now.Add(time.Hour) + query := "" + reason := "" + switch kind { + case "disabled": + query = "UPDATE accounts SET status='disabled'" + reason = "account_inactive" + case "expired": + query = "UPDATE accounts SET expires_at='2026-10-21Z'" + reason = "account_expired" + case "soft_deleted": + query = "UPDATE accounts SET deleted_at=now()" + reason = "account_deleted" + case "hard_deleted": + query = "DELETE FROM accounts" + reason = "account_deleted" + case "late": + *now = now.Add(16 * time.Minute) + reason = "open_expired" + } + if query != "" { + if _, err := s.DB.Exec(query); err != nil { + t.Fatal(err) + } + } + if err := s.RunDue(context.Background()); err != nil { + t.Fatal(err) + } + task := snapshotTask(t, s, "open") + if task.Status != "skipped" || task.Reason != reason || task.ActualAt == nil { + t.Fatal(task) + } + if scalar(t, s.DB, "SELECT count(*) FROM fixture_outbox") != 0 { + t.Fatal("skipped open published account mutation") + } + }) + } +} +func TestIntegrationBoundedRetryAndOutboxRollback(t *testing.T) { + s, now, notified := integrationStore(t) + saveRelative(t, s, "open", 1) + *now = now.Add(time.Hour) + s.Enqueue = func(ctx context.Context, tx *sql.Tx, id int64) error { + _, err := tx.ExecContext(ctx, "INSERT INTO missing_fixture_outbox VALUES($1)", id) + return err + } + for i := 1; i <= 3; i++ { + if err := s.RunDue(context.Background()); err != nil { + t.Fatal(err) + } + task := snapshotTask(t, s, "open") + if task.Attempts != i || task.Reason != "execution_error" { + t.Fatal(task) + } + if scalar(t, s.DB, "SELECT count(*) FROM accounts WHERE schedulable") != 0 { + t.Fatal("account update survived failed outbox transaction") + } + if err := s.RunDue(context.Background()); err != nil { + t.Fatal(err) + } + if snapshotTask(t, s, "open").Attempts != i { + t.Fatal("retry ignored backoff") + } + *now = now.Add(2 * time.Minute) + } + if task := snapshotTask(t, s, "open"); task.Status != "failed" { + t.Fatal(task) + } + if notified.Load() != 0 { + t.Fatal("uncommitted effect notified cache") + } +} +func TestIntegrationChronologicalOrderDuringCompetingClaims(t *testing.T) { + s, now, _ := integrationStore(t) + open := saveRelative(t, s, "open", 1) + closeTask := saveRelative(t, s, "close", 2) + *now = now.Add(2 * time.Hour) + // Force a later claim first. It must wait for the earlier action's terminal + // decision so a delayed opening cannot overwrite a completed close. + if err := s.execute(context.Background(), closeTask); err != nil { + t.Fatal(err) + } + if snapshotTask(t, s, "close").Status != "pending" { + t.Fatal("later action ran out of order") + } + if err := s.execute(context.Background(), open); err != nil { + t.Fatal(err) + } // skips late open + if err := s.execute(context.Background(), closeTask); err != nil { + t.Fatal(err) + } + if snapshotTask(t, s, "close").Status != "executed" || scalar(t, s.DB, "SELECT count(*) FROM accounts WHERE schedulable") != 0 { + t.Fatal("wrong final scheduling state") + } +} +func TestIntegrationManualAndExecutionAreSerialized(t *testing.T) { + s, now, _ := integrationStore(t) + saveRelative(t, s, "open", 1) + *now = now.Add(time.Hour) + entered := make(chan struct{}) + release := make(chan struct{}) + original := s.Enqueue + var once sync.Once + s.Enqueue = func(ctx context.Context, tx *sql.Tx, id int64) error { + once.Do(func() { close(entered); <-release }) + return original(ctx, tx, id) + } + done := make(chan error, 1) + go func() { done <- s.RunDue(context.Background()) }() + select { + case <-entered: + case <-time.After(3 * time.Second): + t.Fatal("worker did not enter transaction") + } + manual := make(chan error, 1) + go func() { manual <- s.Manual(context.Background(), 7, 9, false) }() + close(release) + if err := <-done; err != nil { + t.Fatal(err) + } + if err := <-manual; err != nil { + t.Fatal(err) + } + if scalar(t, s.DB, "SELECT count(*) FROM accounts WHERE schedulable") != 0 { + t.Fatal("worker overwrote later manual close") + } +} +func TestIntegrationAdditiveRollbackFixture(t *testing.T) { + s, _, _ := integrationStore(t) + // Old binaries can use accounts after additive timer tables are removed. + // Real operator rollback must first cancel all plans and stop new workers. + _, err := s.DB.Exec("DROP TABLE account_schedulable_timer_audit; DROP TABLE account_schedulable_timers") + if err != nil { + t.Fatal(err) + } + if scalar(t, s.DB, "SELECT count(*) FROM accounts WHERE status='active'") != 1 { + t.Fatal("existing account schema affected") + } +} +func TestMigrationIsAdditiveAndHasIndexedPendingLookup(t *testing.T) { + b, err := os.ReadFile("../../migrations/295_account_schedulable_timers.sql") + if err != nil { + t.Fatal(err) + } + text := strings.ToLower(string(b)) + for _, required := range []string{"create table if not exists account_schedulable_timers", "primary key (account_id, action)", "on account_schedulable_timers (next_attempt_at, account_id) where status = 'pending'", "create unique index", "actor_id", "timezone", "actual_at"} { + if !strings.Contains(text, required) { + t.Fatalf("missing %s", required) + } + } + for _, forbidden := range []string{"alter table accounts", "drop table", "delete from", "update accounts"} { + if strings.Contains(text, forbidden) { + t.Fatalf("unexpected %s", forbidden) + } + } +} + +func TestIntegrationCancelRacingExecutionHasOneSerializableOutcome(t *testing.T) { + s, now, _ := integrationStore(t) + task := saveRelative(t, s, "open", 1) + *now = now.Add(time.Hour) + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + start := make(chan struct{}) + cancelled := make(chan error, 1) + executed := make(chan error, 1) + go func() { <-start; cancelled <- s.Cancel(ctx, 7, 19, "open", task.Version) }() + go func() { <-start; executed <- s.execute(ctx, task) }() + close(start) + cancelErr := <-cancelled + if err := <-executed; err != nil { + t.Fatal(err) + } + final := snapshotTask(t, s, "open") + effects := scalar(t, s.DB, "SELECT count(*) FROM fixture_outbox") + switch final.Status { + case "cancelled": + if cancelErr != nil || effects != 0 { + t.Fatalf("cancelled but effect survived: %v %d", cancelErr, effects) + } + case "executed": + if ErrorCode(cancelErr) != "version" || effects != 1 { + t.Fatalf("execution/cancel conflict not reported: %v %d", cancelErr, effects) + } + default: + t.Fatal(final) + } + if err := s.execute(ctx, task); err != nil { + t.Fatal(err) + } + if scalar(t, s.DB, "SELECT count(*) FROM fixture_outbox") != effects { + t.Fatal("stale candidate replayed after cancellation race") + } +} +func TestIntegrationEditRacingExecutionFencesTheOldCandidate(t *testing.T) { + s, now, _ := integrationStore(t) + task := saveRelative(t, s, "open", 1) + *now = now.Add(time.Hour) + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + start := make(chan struct{}) + saved := make(chan error, 1) + executed := make(chan error, 1) + go func() { + <-start + _, err := s.Save(ctx, 7, 19, SaveRequest{Input: Input{Mode: "relative", Hours: 2, Timezone: "Asia/Shanghai"}, Action: "open", ExpectedVersion: task.Version}) + saved <- err + }() + go func() { <-start; executed <- s.execute(ctx, task) }() + close(start) + if err := <-saved; err != nil { + t.Fatal(err) + } + if err := <-executed; err != nil { + t.Fatal(err) + } + final := snapshotTask(t, s, "open") + if final.Version != 2 || final.Status != "pending" || !final.ExecuteAt.Equal(now.Add(2*time.Hour)) || final.RequestedBy != 19 { + t.Fatal(final) + } + effects := scalar(t, s.DB, "SELECT count(*) FROM fixture_outbox") + if effects > 1 { + t.Fatal("old version applied multiple times") + } + if err := s.execute(ctx, task); err != nil { + t.Fatal(err) + } + if scalar(t, s.DB, "SELECT count(*) FROM fixture_outbox") != effects { + t.Fatal("old candidate overwrote replacement") + } +} diff --git a/backend/internal/accounttimer/store_test.go b/backend/internal/accounttimer/store_test.go new file mode 100644 index 000000000000..d1a679b8a572 --- /dev/null +++ b/backend/internal/accounttimer/store_test.go @@ -0,0 +1,264 @@ +package accounttimer + +import ( + "context" + "database/sql" + "errors" + "fmt" + "regexp" + "testing" + "time" + + "github.com/DATA-DOG/go-sqlmock" +) + +func testStore(t *testing.T) (*Store, sqlmock.Sqlmock) { + t.Helper() + db, m, err := sqlmock.New() + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = db.Close() }) + s := &Store{DB: db, Now: func() time.Time { return time.Date(2026, 10, 22, 0, 0, 0, 0, time.UTC) }, Policy: DefaultPolicy()} + t.Cleanup(func() { + if err := m.ExpectationsWereMet(); err != nil { + t.Error(err) + } + }) + return s, m +} +func expectAccount(m sqlmock.Sqlmock, status string, deleted bool) { + var d any + if deleted { + d = time.Now() + } + m.ExpectQuery("SELECT status, expires_at, deleted_at FROM accounts").WithArgs(int64(7)).WillReturnRows(sqlmock.NewRows([]string{"status", "expires_at", "deleted_at"}).AddRow(status, nil, d)) +} +func TestSaveRejectsConflictingOppositeAction(t *testing.T) { + s, m := testStore(t) + m.ExpectBegin() + expectAccount(m, "active", false) + m.ExpectQuery("SELECT version FROM account_schedulable_timers").WithArgs(int64(7), "open").WillReturnRows(sqlmock.NewRows([]string{"version"})) + m.ExpectQuery("SELECT EXISTS").WithArgs(int64(7), "open", s.now().Add(8*time.Hour)).WillReturnRows(sqlmock.NewRows([]string{"exists"}).AddRow(true)) + m.ExpectRollback() + _, err := s.Save(context.Background(), 7, 9, SaveRequest{Input: Input{Mode: "relative", Hours: 8, Timezone: "Asia/Shanghai"}, Action: "open"}) + if ErrorCode(err) != "conflict" { + t.Fatalf("want conflict, got %v", err) + } +} +func TestSaveRejectsStaleEdit(t *testing.T) { + s, m := testStore(t) + m.ExpectBegin() + expectAccount(m, "active", false) + m.ExpectQuery("SELECT version FROM account_schedulable_timers").WithArgs(int64(7), "open").WillReturnRows(sqlmock.NewRows([]string{"version"}).AddRow(4)) + m.ExpectRollback() + _, err := s.Save(context.Background(), 7, 9, SaveRequest{Input: Input{Mode: "relative", Hours: 8, Timezone: "UTC"}, Action: "open", ExpectedVersion: 3}) + if ErrorCode(err) != "version" { + t.Fatalf("want version, got %v", err) + } +} +func TestSaveRequiresFixedPreviewConfirmation(t *testing.T) { + s, m := testStore(t) + m.ExpectBegin() + expectAccount(m, "active", false) + m.ExpectRollback() + _, err := s.Save(context.Background(), 7, 9, SaveRequest{Input: Input{Mode: "absolute", Time: "08:00", Timezone: "Asia/Shanghai"}, Action: "open"}) + if ErrorCode(err) != "confirmation" { + t.Fatalf("want confirmation, got %v", err) + } +} +func TestCancelVersionChecked(t *testing.T) { + s, m := testStore(t) + m.ExpectBegin() + expectAccount(m, "active", false) + m.ExpectQuery("SELECT "+regexp.QuoteMeta(taskColumns)).WithArgs(int64(7), "open").WillReturnRows(taskRows().AddRow(7, "open", 4, s.now().Add(time.Hour), "UTC", 9, "pending", 0, nil, "")) + m.ExpectRollback() + if err := s.Cancel(context.Background(), 7, 9, "open", 3); ErrorCode(err) != "version" { + t.Fatalf("want version, got %v", err) + } +} +func taskRows() *sqlmock.Rows { + return sqlmock.NewRows([]string{"account_id", "action", "version", "execute_at", "timezone", "requested_by", "status", "attempts", "actual_at", "reason"}) +} + +// The outbox is a transactional dependency; a failure must roll back schedulable +// and the terminal result, leaving a bounded retry instead of a lost event. +func TestExecuteOutboxFailureRollsBackAndRecordsRetry(t *testing.T) { + s, m := testStore(t) + s.Enqueue = func(context.Context, *sql.Tx, int64) error { return errors.New("fixture outbox unavailable") } + now := s.now() + task := Task{AccountID: 7, Action: "open", Version: 1, ExecuteAt: now, Timezone: "UTC", RequestedBy: 9, Status: "pending"} + m.ExpectBegin() + expectAccount(m, "active", false) + m.ExpectQuery("SELECT "+regexp.QuoteMeta(taskColumns)).WithArgs(int64(7), "open", now).WillReturnRows(taskRows().AddRow(7, "open", 1, now, "UTC", 9, "pending", 0, nil, "")) + m.ExpectExec("UPDATE accounts SET schedulable").WithArgs(true, now, int64(7)).WillReturnResult(sqlmock.NewResult(0, 1)) + m.ExpectRollback() + m.ExpectBegin() + expectAccount(m, "active", false) + m.ExpectExec("UPDATE account_schedulable_timers").WillReturnResult(sqlmock.NewResult(0, 1)) + m.ExpectExec("INSERT INTO account_schedulable_timer_audit").WillReturnResult(sqlmock.NewResult(1, 1)) + m.ExpectCommit() + if err := s.execute(context.Background(), task); err != nil { + t.Fatal(err) + } +} + +func TestSavePersistsUTCAndOriginalZoneAndReplacesSameAction(t *testing.T) { + for _, version := range []int64{0, 3} { + t.Run(fmt.Sprint(version), func(t *testing.T) { + s, m := testStore(t) + now := s.now() + due := now.Add(8 * time.Hour) + m.ExpectBegin() + expectAccount(m, "active", false) + rows := sqlmock.NewRows([]string{"version"}) + if version > 0 { + rows.AddRow(version) + } + m.ExpectQuery("SELECT version FROM account_schedulable_timers").WithArgs(int64(7), "open").WillReturnRows(rows) + m.ExpectQuery("SELECT EXISTS").WithArgs(int64(7), "open", due).WillReturnRows(sqlmock.NewRows([]string{"exists"}).AddRow(false)) + m.ExpectExec("INSERT INTO account_schedulable_timers").WithArgs(int64(7), "open", version+1, due, "Asia/Shanghai", int64(9)).WillReturnResult(sqlmock.NewResult(0, 1)) + result := "scheduled" + if version > 0 { + result = "replaced" + } + m.ExpectExec("INSERT INTO account_schedulable_timer_audit").WithArgs(int64(7), "open", version+1, int64(9), due, "Asia/Shanghai", now, result, "").WillReturnResult(sqlmock.NewResult(1, 1)) + m.ExpectCommit() + task, err := s.Save(context.Background(), 7, 9, SaveRequest{Input: Input{Mode: "relative", Hours: 8, Timezone: "Asia/Shanghai"}, Action: "open", ExpectedVersion: version}) + if err != nil || !task.ExecuteAt.Equal(due) || task.Version != version+1 || task.Timezone != "Asia/Shanghai" { + t.Fatalf("%+v %v", task, err) + } + }) + } +} +func TestManualOverrideCancelsBothActionsAndPublishesOnlyAfterCommit(t *testing.T) { + s, m := testStore(t) + now := s.now() + notified := 0 + s.Enqueue = func(ctx context.Context, tx *sql.Tx, id int64) error { + _, err := tx.ExecContext(ctx, "INSERT INTO fixture_outbox VALUES ($1)", id) + return err + } + s.AfterCommit = func(context.Context, int64) { notified++ } + m.ExpectBegin() + expectAccount(m, "active", false) + m.ExpectExec("WITH cancelled AS").WithArgs(int64(7), int64(9), now).WillReturnResult(sqlmock.NewResult(0, 2)) + m.ExpectExec("UPDATE accounts SET schedulable").WithArgs(false, now, int64(7)).WillReturnResult(sqlmock.NewResult(0, 1)) + m.ExpectExec("INSERT INTO fixture_outbox").WithArgs(int64(7)).WillReturnResult(sqlmock.NewResult(0, 1)) + m.ExpectCommit() + if err := s.Manual(context.Background(), 7, 9, false); err != nil { + t.Fatal(err) + } + if notified != 1 { + t.Fatal("cache update missing") + } +} +func TestManualOutboxFailureRollsBackCancellationAndSwitch(t *testing.T) { + s, m := testStore(t) + s.Enqueue = func(context.Context, *sql.Tx, int64) error { return errors.New("fixture failure") } + s.AfterCommit = func(context.Context, int64) { t.Error("cache must not change before commit") } + m.ExpectBegin() + expectAccount(m, "active", false) + m.ExpectExec("WITH cancelled AS").WillReturnResult(sqlmock.NewResult(0, 2)) + m.ExpectExec("UPDATE accounts SET schedulable").WillReturnResult(sqlmock.NewResult(0, 1)) + m.ExpectRollback() + if err := s.Manual(context.Background(), 7, 9, false); err == nil { + t.Fatal("must reject partial update") + } +} +func TestAutomaticPoliciesDoNotCancelPlans(t *testing.T) { + s, _ := testStore(t) + if err := CancelPending(context.Background(), s.DB, 7, s.now()); err != nil { + t.Fatal(err) + } + ctx := WithManualAccount(WithManualIntent(context.Background(), 9), 8) + if err := CancelPending(ctx, s.DB, 7, s.now()); err != nil { + t.Fatal(err) + } +} +func TestExecutionRechecksVersionStatusAndNotifiesAfterCommit(t *testing.T) { + cases := []struct { + name, action, status, accountStatus, reason string + version int64 + deleted bool + }{ + {"restart opens once", "open", "pending", "active", "", 1, false}, + {"stale worker cannot overwrite edit", "open", "pending", "active", "", 2, false}, + {"second instance sees executed", "open", "executed", "active", "", 1, false}, + {"deleted account skipped", "open", "pending", "active", "account_deleted", 1, true}, + {"disabled remains disabled", "open", "pending", "disabled", "account_inactive", 1, false}, + {"offline close catchup", "close", "pending", "active", "", 1, false}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + s, m := testStore(t) + now := s.now() + due := now + if c.action == "close" { + due = now.Add(-24 * time.Hour) + } + candidate := Task{AccountID: 7, Action: c.action, Version: 1, ExecuteAt: due, Timezone: "Asia/Shanghai", RequestedBy: 9, Status: "pending"} + notified := 0 + s.AfterCommit = func(context.Context, int64) { notified++ } + m.ExpectBegin() + expectAccount(m, c.accountStatus, c.deleted) + m.ExpectQuery("SELECT "+regexp.QuoteMeta(taskColumns)).WithArgs(int64(7), c.action, now).WillReturnRows(taskRows().AddRow(7, c.action, c.version, due, "Asia/Shanghai", 9, c.status, 0, nil, "")) + active := c.version == 1 && c.status == "pending" + result := "executed" + if c.reason != "" { + result = "skipped" + } + if active { + if c.reason == "" { + m.ExpectExec("UPDATE accounts SET schedulable").WithArgs(c.action == "open", now, int64(7)).WillReturnResult(sqlmock.NewResult(0, 1)) + } + m.ExpectExec("UPDATE account_schedulable_timers").WithArgs(int64(7), c.action, result, now, c.reason, int64(1)).WillReturnResult(sqlmock.NewResult(0, 1)) + m.ExpectExec("INSERT INTO account_schedulable_timer_audit").WithArgs(int64(7), c.action, int64(1), int64(9), due, "Asia/Shanghai", now, result, c.reason).WillReturnResult(sqlmock.NewResult(1, 1)) + m.ExpectCommit() + } else { + m.ExpectRollback() + } + if err := s.execute(context.Background(), candidate); err != nil { + t.Fatal(err) + } + want := 0 + if active && c.reason == "" { + want = 1 + } + if notified != want { + t.Fatalf("cache calls %d want %d", notified, want) + } + }) + } +} +func TestThirdFailureBecomesTerminalAndStaleRetryCannotChangeNewVersion(t *testing.T) { + for _, affected := range []int64{1, 0} { + t.Run(fmt.Sprint(affected), func(t *testing.T) { + s, m := testStore(t) + now := s.now() + task := Task{AccountID: 7, Action: "open", Version: 1, ExecuteAt: now, Timezone: "UTC", RequestedBy: 9, Attempts: 2} + m.ExpectBegin() + expectAccount(m, "active", false) + m.ExpectExec("UPDATE account_schedulable_timers").WithArgs(int64(7), "open", int64(1), 3, "failed", now.Add(120*time.Second), now, 2).WillReturnResult(sqlmock.NewResult(0, affected)) + if affected == 1 { + m.ExpectExec("INSERT INTO account_schedulable_timer_audit").WillReturnResult(sqlmock.NewResult(1, 1)) + m.ExpectCommit() + } else { + m.ExpectRollback() + } + if err := s.retry(context.Background(), task); err != nil { + t.Fatal(err) + } + }) + } +} +func TestBusyAccountIsSkippedWithoutMisreportingDeletion(t *testing.T) { + s, m := testStore(t) + m.ExpectBegin() + m.ExpectQuery("SELECT status, expires_at, deleted_at FROM accounts").WillReturnRows(sqlmock.NewRows([]string{"status", "expires_at", "deleted_at"})) + m.ExpectQuery("SELECT EXISTS").WithArgs(int64(7)).WillReturnRows(sqlmock.NewRows([]string{"exists"}).AddRow(true)) + m.ExpectRollback() + if err := s.execute(context.Background(), Task{AccountID: 7, Action: "open", Version: 1}); err != nil { + t.Fatal(err) + } +} diff --git a/backend/internal/accounttimer/time.go b/backend/internal/accounttimer/time.go new file mode 100644 index 000000000000..7d7d614827ac --- /dev/null +++ b/backend/internal/accounttimer/time.go @@ -0,0 +1,157 @@ +// Package accounttimer implements one-shot scheduling without model calls or browser timers. +package accounttimer + +import ( + "errors" + "os" + "strconv" + "strings" + "time" + _ "time/tzdata" +) + +type Error struct{ Code string } + +func (e *Error) Error() string { return "account timer: " + e.Code } +func ErrorCode(err error) string { + var e *Error + if errors.As(err, &e) { + return e.Code + } + return "internal" +} +func invalid(code string) error { return &Error{Code: code} } + +type Policy struct { + MaxDelay time.Duration + OpenGrace time.Duration +} + +func DefaultPolicy() Policy { return Policy{366 * 24 * time.Hour, 15 * time.Minute} } + +const PollInterval = 30 * time.Second + +type Input struct { + Mode string `json:"mode"` + Hours int64 `json:"hours"` + Minutes int64 `json:"minutes"` + Date string `json:"date"` + Time string `json:"time"` + Timezone string `json:"timezone"` +} +type Preview struct { + ExecuteAt time.Time `json:"execute_at"` + Timezone string `json:"timezone"` + LocalDate string `json:"local_date"` + LocalTime string `json:"local_time"` + Offset string `json:"offset"` + ServerNow time.Time `json:"server_now"` +} + +// ConfiguredPolicy bounds the operator-configurable planning horizon. Invalid values +// fail closed to the documented default; the maximum remains ten years. +func ConfiguredPolicy() Policy { + p := DefaultPolicy() + if days, err := strconv.Atoi(os.Getenv("ACCOUNT_SCHEDULABLE_TIMER_MAX_DAYS")); err == nil && days >= 1 && days <= 3660 { + p.MaxDelay = time.Duration(days) * 24 * time.Hour + } + return p +} +func Resolve(in Input, now time.Time, p Policy) (Preview, error) { + var out Preview + if in.Timezone == "" || in.Timezone == "Local" || (in.Timezone != "UTC" && !strings.Contains(in.Timezone, "/")) { + return out, invalid("timezone") + } + loc, err := time.LoadLocation(in.Timezone) + if err != nil { + return out, invalid("timezone") + } + var due time.Time + switch in.Mode { + case "relative": + maxMinutes := int64(p.MaxDelay / time.Minute) + if in.Date != "" || in.Time != "" || in.Hours < 0 || in.Minutes < 0 || in.Minutes > maxMinutes || in.Hours > maxMinutes/60 { + return out, invalid("duration") + } + minutes := in.Hours*60 + in.Minutes + if minutes <= 0 || minutes > maxMinutes { + return out, invalid("duration") + } + due = now.Add(time.Duration(minutes) * time.Minute) + case "absolute": + if in.Hours != 0 || in.Minutes != 0 { + return out, invalid("datetime") + } + date := in.Date + if date == "" { + date = now.In(loc).Format("2006-01-02") + } + due, err = resolveWall(date, in.Time, loc) + if err != nil { + return out, err + } + if in.Date == "" && !due.After(now) { + day, _ := time.Parse("2006-01-02", date) + due, err = resolveWall(day.AddDate(0, 0, 1).Format("2006-01-02"), in.Time, loc) + if err != nil { + return out, err + } + } + default: + return out, invalid("mode") + } + if !due.After(now) { + return out, invalid("past") + } + if due.Sub(now) > p.MaxDelay { + return out, invalid("duration") + } + local := due.In(loc) + return Preview{due.UTC(), in.Timezone, local.Format("2006-01-02"), local.Format("15:04"), local.Format("-07:00"), now.UTC()}, nil +} + +// Enumerate offsets around the requested wall date, then round-trip every +// candidate. time.Date alone silently normalizes gaps and chooses one fold. +func resolveWall(date, clock string, loc *time.Location) (time.Time, error) { + text := date + " " + clock + wall, err := time.Parse("2006-01-02 15:04", text) + if err != nil || wall.Format("2006-01-02 15:04") != text { + return time.Time{}, invalid("datetime") + } + offsets := map[int]bool{} + for t := wall.Add(-48 * time.Hour); !t.After(wall.Add(48 * time.Hour)); t = t.Add(30 * time.Minute) { + _, off := t.In(loc).Zone() + offsets[off] = true + } + matches := []time.Time{} + for off := range offsets { + candidate := wall.Add(-time.Duration(off) * time.Second) + if candidate.In(loc).Format("2006-01-02 15:04") == text { + matches = append(matches, candidate) + } + } + if len(matches) != 1 { + return time.Time{}, invalid("dst") + } + return matches[0], nil +} + +// Opening changes only schedulable, and never makes an inactive/expired account +// eligible. Quota, cooldown, ACL and quality gates remain untouched. +func Decision(open bool, due, now time.Time, status string, expires *time.Time, deleted bool, p Policy) string { + if deleted { + return "account_deleted" + } + if open { + if now.Sub(due) > p.OpenGrace { + return "open_expired" + } + if status != "active" { + return "account_inactive" + } + if expires != nil && !expires.After(now) { + return "account_expired" + } + } + return "" +} diff --git a/backend/internal/accounttimer/time_test.go b/backend/internal/accounttimer/time_test.go new file mode 100644 index 000000000000..3808396af9f1 --- /dev/null +++ b/backend/internal/accounttimer/time_test.go @@ -0,0 +1,74 @@ +package accounttimer + +import ( + "testing" + "time" +) + +func TestResolve(t *testing.T) { + now := time.Date(2026, 10, 21, 23, 30, 0, 0, time.UTC) + cases := []struct { + name string + input Input + want string + code string + }{ + {"8h from server receipt", Input{Mode: "relative", Hours: 8, Timezone: "Asia/Shanghai"}, "2026-10-22T07:30:00Z", ""}, + {"minutes cross midnight", Input{Mode: "relative", Minutes: 45, Timezone: "Asia/Shanghai"}, "2026-10-22T00:15:00Z", ""}, + {"Beijing next 8", Input{Mode: "absolute", Time: "08:00", Timezone: "Asia/Shanghai"}, "2026-10-22T00:00:00Z", ""}, + {"explicit past stays past", Input{Mode: "absolute", Date: "2026-10-21", Time: "08:00", Timezone: "Asia/Shanghai"}, "", "past"}, + {"explicit Beijing date", Input{Mode: "absolute", Date: "2026-10-23", Time: "08:00", Timezone: "Asia/Shanghai"}, "2026-10-23T00:00:00Z", ""}, + {"invalid zone", Input{Mode: "relative", Hours: 8, Timezone: "Mars/City"}, "", "timezone"}, + {"no implicit zone", Input{Mode: "relative", Hours: 8}, "", "timezone"}, + {"DST missing", Input{Mode: "absolute", Date: "2027-03-14", Time: "02:30", Timezone: "America/New_York"}, "", "dst"}, + {"DST duplicate", Input{Mode: "absolute", Date: "2026-11-01", Time: "01:30", Timezone: "America/New_York"}, "", "dst"}, + {"zero", Input{Mode: "relative", Timezone: "UTC"}, "", "duration"}, + {"extreme", Input{Mode: "relative", Hours: 1 << 60, Timezone: "UTC"}, "", "duration"}, + {"negative", Input{Mode: "relative", Hours: -1, Minutes: 90, Timezone: "UTC"}, "", "duration"}, + {"invalid calendar date", Input{Mode: "absolute", Date: "2026-02-30", Time: "08:00", Timezone: "UTC"}, "", "datetime"}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + got, err := Resolve(c.input, now, DefaultPolicy()) + if c.code != "" { + if ErrorCode(err) != c.code { + t.Fatalf("want %s, got %v", c.code, err) + } + return + } + if err != nil || got.ExecuteAt.Format(time.RFC3339) != c.want { + t.Fatalf("got %+v %v want %s", got, err, c.want) + } + if got.Timezone != c.input.Timezone || got.LocalDate == "" || got.Offset == "" { + t.Fatal("missing explicit zone preview") + } + }) + } +} +func TestNextEightCrossesLocalDay(t *testing.T) { + now := time.Date(2026, 10, 22, 0, 0, 0, 0, time.UTC) + p, err := Resolve(Input{Mode: "absolute", Time: "08:00", Timezone: "Asia/Shanghai"}, now, DefaultPolicy()) + if err != nil || p.ExecuteAt.Format(time.RFC3339) != "2026-10-23T00:00:00Z" { + t.Fatalf("%+v %v", p, err) + } +} +func TestExecutionPolicy(t *testing.T) { + now := time.Now() + p := DefaultPolicy() + if Decision(true, now.Add(-16*time.Minute), now, "active", nil, false, p) != "open_expired" { + t.Fatal("late open must expire") + } + if Decision(false, now.Add(-365*24*time.Hour), now, "active", nil, false, p) != "" { + t.Fatal("late close must run") + } + if Decision(true, now, now, "disabled", nil, false, p) != "account_inactive" { + t.Fatal("must preserve disabled") + } + expired := now.Add(-time.Hour) + if Decision(true, now, now, "active", &expired, false, p) != "account_expired" { + t.Fatal("must preserve expiry") + } + if Decision(false, now, now, "active", nil, true, p) != "account_deleted" { + t.Fatal("must preserve deletion") + } +} diff --git a/backend/internal/handler/admin/account_handler.go b/backend/internal/handler/admin/account_handler.go index 01f916b82c35..bbe0e674594e 100644 --- a/backend/internal/handler/admin/account_handler.go +++ b/backend/internal/handler/admin/account_handler.go @@ -1247,7 +1247,7 @@ func (h *AccountHandler) Update(c *gin.Context) { // 确定是否跳过混合渠道检查 skipCheck := req.ConfirmMixedChannelRisk != nil && *req.ConfirmMixedChannelRisk - account, err := h.adminService.UpdateAccount(c.Request.Context(), accountID, &service.UpdateAccountInput{ + account, err := h.adminService.UpdateAccount(accountManualSchedulingContext(c, req.Schedulable != nil), accountID, &service.UpdateAccountInput{ Name: req.Name, Notes: req.Notes, Type: req.Type, @@ -2477,7 +2477,7 @@ func (h *AccountHandler) BulkUpdate(c *gin.Context) { return } - result, err := h.adminService.BulkUpdateAccounts(c.Request.Context(), &service.BulkUpdateAccountsInput{ + result, err := h.adminService.BulkUpdateAccounts(accountManualSchedulingContext(c, req.Schedulable != nil), &service.BulkUpdateAccountsInput{ AccountIDs: req.AccountIDs, Filters: toServiceBulkUpdateAccountFilters(req.Filters), Name: req.Name, @@ -2915,7 +2915,7 @@ func (h *AccountHandler) SetSchedulable(c *gin.Context) { return } - account, err := h.adminService.SetAccountSchedulable(c.Request.Context(), accountID, req.Schedulable) + account, err := h.adminService.SetAccountSchedulable(accountManualSchedulingContext(c, true), accountID, req.Schedulable) if err != nil { response.ErrorFrom(c, err) return diff --git a/backend/internal/handler/admin/account_schedulable_timer.go b/backend/internal/handler/admin/account_schedulable_timer.go new file mode 100644 index 000000000000..9d711df95594 --- /dev/null +++ b/backend/internal/handler/admin/account_schedulable_timer.go @@ -0,0 +1,41 @@ +package admin + +import ( + "context" + + "github.com/Wei-Shaw/sub2api/internal/accounttimer" + "github.com/Wei-Shaw/sub2api/internal/handler/admin/accounttimerapi" + "github.com/Wei-Shaw/sub2api/internal/server/middleware" + "github.com/gin-gonic/gin" +) + +func accountManualSchedulingContext(c *gin.Context, explicit bool) context.Context { + ctx := c.Request.Context() + if explicit { + if subject, ok := middleware.GetAuthSubjectFromContext(c); ok && subject.UserID > 0 { + return accounttimer.WithManualIntent(ctx, subject.UserID) + } + } + return ctx +} +func (h *AccountHandler) schedulableTimerAPI() *accounttimerapi.Handler { + return &accounttimerapi.Handler{ + Repository: func() accounttimer.Repository { + if provider, ok := h.adminService.(interface { + SchedulableTimers() accounttimer.Repository + }); ok { + return provider.SchedulableTimers() + } + return nil + }, + Identity: func(c *gin.Context) (int64, string) { + subject, _ := middleware.GetAuthSubjectFromContext(c) + role, _ := middleware.GetUserRoleFromContext(c) + return subject.UserID, role + }, + } +} +func (h *AccountHandler) ListSchedulableTimers(c *gin.Context) { h.schedulableTimerAPI().List(c) } +func (h *AccountHandler) PreviewSchedulableTimer(c *gin.Context) { h.schedulableTimerAPI().Preview(c) } +func (h *AccountHandler) SaveSchedulableTimer(c *gin.Context) { h.schedulableTimerAPI().Save(c) } +func (h *AccountHandler) CancelSchedulableTimer(c *gin.Context) { h.schedulableTimerAPI().Cancel(c) } diff --git a/backend/internal/handler/admin/accounttimerapi/handler.go b/backend/internal/handler/admin/accounttimerapi/handler.go new file mode 100644 index 000000000000..e8582512a02e --- /dev/null +++ b/backend/internal/handler/admin/accounttimerapi/handler.go @@ -0,0 +1,121 @@ +// Package accounttimerapi keeps the admin timer API independently testable. +package accounttimerapi + +import ( + "context" + "errors" + "net/http" + "strconv" + "time" + + "github.com/Wei-Shaw/sub2api/internal/accounttimer" + "github.com/Wei-Shaw/sub2api/internal/pkg/response" + "github.com/gin-gonic/gin" +) + +type Handler struct { + Repository func() accounttimer.Repository + Identity func(*gin.Context) (int64, string) +} + +func (h *Handler) access(c *gin.Context) (accounttimer.Repository, int64, int64, bool) { + actor, role := h.Identity(c) + if actor <= 0 || role != "admin" { + response.Forbidden(c, "Administrator access required") + return nil, 0, 0, false + } + id, err := strconv.ParseInt(c.Param("id"), 10, 64) + if err != nil || id <= 0 { + response.BadRequest(c, "Invalid account ID") + return nil, 0, 0, false + } + repo := h.Repository() + if repo == nil { + response.Error(c, http.StatusServiceUnavailable, "Account timers unavailable") + return nil, 0, 0, false + } + return repo, id, actor, true +} +func accountTimerError(c *gin.Context, err error) { + var typed *accounttimer.Error + if !errors.As(err, &typed) { + response.ErrorWithDetails(c, http.StatusInternalServerError, "Account timer operation failed", "TIMER_internal", nil) + return + } + status := http.StatusBadRequest + if typed.Code == "version" || typed.Code == "conflict" { + status = http.StatusConflict + } + if typed.Code == "account_deleted" { + status = http.StatusNotFound + } + response.ErrorWithDetails(c, status, "Account timer: "+typed.Code, "TIMER_"+typed.Code, nil) +} +func (h *Handler) List(c *gin.Context) { + repo, id, _, ok := h.access(c) + if !ok { + return + } + ctx, cancel := context.WithTimeout(c.Request.Context(), 5*time.Second) + defer cancel() + result, err := repo.List(ctx, id) + if err != nil { + accountTimerError(c, err) + return + } + response.Success(c, result) +} +func (h *Handler) Preview(c *gin.Context) { + repo, _, _, ok := h.access(c) + if !ok { + return + } + var req accounttimer.Input + if c.ShouldBindJSON(&req) != nil { + response.BadRequest(c, "Invalid timer request") + return + } + preview, err := repo.Preview(req) + if err != nil { + accountTimerError(c, err) + return + } + response.Success(c, preview) +} +func (h *Handler) Save(c *gin.Context) { + repo, id, actor, ok := h.access(c) + if !ok { + return + } + var req accounttimer.SaveRequest + if c.ShouldBindJSON(&req) != nil { + response.BadRequest(c, "Invalid timer request") + return + } + ctx, cancel := context.WithTimeout(c.Request.Context(), 5*time.Second) + defer cancel() + task, err := repo.Save(ctx, id, actor, req) + if err != nil { + accountTimerError(c, err) + return + } + response.Success(c, task) +} +func (h *Handler) Cancel(c *gin.Context) { + repo, id, actor, ok := h.access(c) + if !ok { + return + } + version, err := strconv.ParseInt(c.Query("version"), 10, 64) + if err != nil { + response.BadRequest(c, "Invalid timer version") + return + } + ctx, cancel := context.WithTimeout(c.Request.Context(), 5*time.Second) + defer cancel() + if err = repo.Cancel(ctx, id, actor, c.Param("action"), version); err != nil { + accountTimerError(c, err) + return + } + response.Success(c, gin.H{"cancelled": true}) +} diff --git a/backend/internal/handler/admin/accounttimerapi/handler_test.go b/backend/internal/handler/admin/accounttimerapi/handler_test.go new file mode 100644 index 000000000000..c373bfc7e7a8 --- /dev/null +++ b/backend/internal/handler/admin/accounttimerapi/handler_test.go @@ -0,0 +1,129 @@ +package accounttimerapi + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/Wei-Shaw/sub2api/internal/accounttimer" + "github.com/gin-gonic/gin" +) + +type fakeRepo struct { + accounttimer.Repository + calls int + id, actor, version int64 + action string + err error +} + +func (r *fakeRepo) List(_ context.Context, id int64) (accounttimer.Snapshot, error) { + r.calls++ + r.id = id + return accounttimer.Snapshot{Tasks: []accounttimer.Task{}}, r.err +} +func (r *fakeRepo) Preview(in accounttimer.Input) (accounttimer.Preview, error) { + r.calls++ + return accounttimer.Resolve(in, time.Date(2026, 10, 22, 0, 0, 0, 0, time.UTC), accounttimer.DefaultPolicy()) +} +func (r *fakeRepo) Save(ctx context.Context, id, actor int64, req accounttimer.SaveRequest) (accounttimer.Task, error) { + r.calls++ + r.id = id + r.actor = actor + r.action = req.Action + r.version = req.ExpectedVersion + if _, ok := ctx.Deadline(); !ok { + panic("unbounded admin transaction") + } + return accounttimer.Task{AccountID: id, Action: req.Action}, r.err +} +func (r *fakeRepo) Cancel(_ context.Context, id, actor int64, action string, version int64) error { + r.calls++ + r.id = id + r.actor = actor + r.action = action + r.version = version + return r.err +} +func router(repo *fakeRepo, role string, actor int64) *gin.Engine { + gin.SetMode(gin.TestMode) + r := gin.New() + h := &Handler{Repository: func() accounttimer.Repository { return repo }, Identity: func(*gin.Context) (int64, string) { return actor, role }} + r.GET("/accounts/:id/timers", h.List) + r.POST("/accounts/:id/timers/preview", h.Preview) + r.POST("/accounts/:id/timers", h.Save) + r.DELETE("/accounts/:id/timers/:action", h.Cancel) + return r +} +func TestAllTimerAPIsRejectNonAdminBeforeAccess(t *testing.T) { + endpoints := []struct{ method, path string }{{"GET", "/accounts/7/timers"}, {"POST", "/accounts/7/timers/preview"}, {"POST", "/accounts/7/timers"}, {"DELETE", "/accounts/7/timers/open?version=1"}} + for _, role := range []string{"", "user", "observer"} { + for _, ep := range endpoints { + t.Run(role+ep.method+ep.path, func(t *testing.T) { + repo := &fakeRepo{} + w := httptest.NewRecorder() + router(repo, role, 9).ServeHTTP(w, httptest.NewRequest(ep.method, ep.path, strings.NewReader(`{}`))) + if w.Code != http.StatusForbidden || repo.calls != 0 { + t.Fatalf("status %d calls %d", w.Code, repo.calls) + } + }) + } + } +} +func TestTimerAPIPropagatesActorAndVersion(t *testing.T) { + repo := &fakeRepo{} + r := router(repo, "admin", 19) + w := httptest.NewRecorder() + r.ServeHTTP(w, httptest.NewRequest("POST", "/accounts/7/timers", strings.NewReader(`{"mode":"relative","hours":8,"timezone":"Asia/Shanghai","action":"open","expected_version":3}`))) + if w.Code != 200 || repo.id != 7 || repo.actor != 19 || repo.version != 3 || repo.action != "open" { + t.Fatalf("%d %+v", w.Code, repo) + } + w = httptest.NewRecorder() + r.ServeHTTP(w, httptest.NewRequest("DELETE", "/accounts/7/timers/close?version=4", nil)) + if w.Code != 200 || repo.actor != 19 || repo.version != 4 || repo.action != "close" { + t.Fatalf("%d %+v", w.Code, repo) + } +} +func TestTimerAPIRejectsMalformedIDsBodiesAndVersions(t *testing.T) { + for _, ep := range []struct{ method, path, body string }{{"GET", "/accounts/0/timers", ""}, {"POST", "/accounts/x/timers", "{}"}, {"POST", "/accounts/7/timers", "{"}, {"DELETE", "/accounts/7/timers/open?version=x", ""}} { + repo := &fakeRepo{} + w := httptest.NewRecorder() + router(repo, "admin", 9).ServeHTTP(w, httptest.NewRequest(ep.method, ep.path, strings.NewReader(ep.body))) + if w.Code != 400 || repo.calls != 0 { + t.Fatalf("%s status %d calls %d", ep.path, w.Code, repo.calls) + } + } +} +func TestTimerAPIErrorsAreStructuredAndRedacted(t *testing.T) { + for _, code := range []string{"conflict", "version", "account_deleted", "dst"} { + repo := &fakeRepo{err: &accounttimer.Error{Code: code}} + w := httptest.NewRecorder() + router(repo, "admin", 9).ServeHTTP(w, httptest.NewRequest("POST", "/accounts/7/timers", strings.NewReader(`{"action":"open"}`))) + want := 400 + if code == "conflict" || code == "version" { + want = 409 + } + if code == "account_deleted" { + want = 404 + } + var body struct{ Reason string } + if err := json.Unmarshal(w.Body.Bytes(), &body); err != nil { + t.Fatal(err) + } + if w.Code != want || body.Reason != "TIMER_"+code { + t.Fatalf("%d %s", w.Code, w.Body.String()) + } + } +} +func TestPreviewNextBeijingEightReturnsConcreteDateAndOffset(t *testing.T) { + repo := &fakeRepo{} + w := httptest.NewRecorder() + router(repo, "admin", 9).ServeHTTP(w, httptest.NewRequest("POST", "/accounts/7/timers/preview", strings.NewReader(`{"mode":"absolute","time":"08:00","timezone":"Asia/Shanghai"}`))) + if w.Code != 200 || !strings.Contains(w.Body.String(), `"local_date":"2026-10-23"`) || !strings.Contains(w.Body.String(), `"offset":"+08:00"`) { + t.Fatalf("%d %s", w.Code, w.Body.String()) + } +} diff --git a/backend/internal/repository/account_repo.go b/backend/internal/repository/account_repo.go index 15d19d9c14d0..e5fb3616c80c 100644 --- a/backend/internal/repository/account_repo.go +++ b/backend/internal/repository/account_repo.go @@ -27,6 +27,7 @@ import ( dbgroup "github.com/Wei-Shaw/sub2api/ent/group" dbpredicate "github.com/Wei-Shaw/sub2api/ent/predicate" dbproxy "github.com/Wei-Shaw/sub2api/ent/proxy" + "github.com/Wei-Shaw/sub2api/internal/accounttimer" "github.com/Wei-Shaw/sub2api/internal/pkg/logger" "github.com/Wei-Shaw/sub2api/internal/pkg/pagination" "github.com/Wei-Shaw/sub2api/internal/service" @@ -581,6 +582,7 @@ func (r *accountRepository) updateAccount( } account.UpdatedAt = updated.UpdatedAt + account.Schedulable = updated.Schedulable // 普通账号编辑（如 model_mapping / credentials）也需要立即刷新单账号快照， // 否则网关在 outbox worker 延迟或异常时仍可能读到旧配置。 if contextTx == nil { @@ -602,6 +604,10 @@ func (r *accountRepository) updateLockedAccount( return nil, err } account.Extra = extra + // lockAndMergeAccountProbeExtra holds the account row lock in this transaction. + if err := accounttimer.CancelPending(ctx, client, account.ID, time.Now().UTC()); err != nil { + return nil, err + } schedulable := account.Schedulable if account.Status == service.StatusError { @@ -619,8 +625,10 @@ func (r *accountRepository) updateLockedAccount( SetPriority(account.Priority). SetStatus(account.Status). SetErrorMessage(account.ErrorMessage). - SetSchedulable(schedulable). SetAutoPauseOnExpired(account.AutoPauseOnExpired) + if !accounttimer.PreserveSchedulable(ctx) || account.Status == service.StatusError { + builder.SetSchedulable(schedulable) + } if explicitRateMultiplier != nil { builder.SetRateMultiplier(*explicitRateMultiplier) @@ -3036,6 +3044,17 @@ func (r *accountRepository) UpdateSessionWindowEnd(ctx context.Context, id int64 } func (r *accountRepository) SetSchedulable(ctx context.Context, id int64, schedulable bool) error { + if actor := accounttimer.ManualActor(ctx); actor > 0 { + store, ok := r.SchedulableTimers().(*accounttimer.Store) + if !ok { + return errors.New("account timer transaction unavailable") + } + err := store.Manual(ctx, id, actor, schedulable) + if accounttimer.ErrorCode(err) == "account_deleted" { + return service.ErrAccountNotFound + } + return err + } _, err := r.client.Account.Update(). Where(dbaccount.IDEQ(id)). SetSchedulable(schedulable). @@ -3666,6 +3685,34 @@ func (r *accountRepository) BulkUpdate(ctx context.Context, ids []int64, updates } } + if updates.Schedulable != nil && accounttimer.ManualActor(ctx) > 0 { + if contextTx == nil && tx == nil { + return 0, errors.New("manual bulk scheduling requires a transaction") + } + locked, err := exec.QueryContext(ctx, "SELECT id FROM accounts WHERE id=ANY($1) AND deleted_at IS NULL ORDER BY id FOR UPDATE", pq.Array(ids)) + if err != nil { + return 0, err + } + lockedIDs := []int64{} + for locked.Next() { + var id int64 + if err = locked.Scan(&id); err != nil { + _ = locked.Close() + return 0, err + } + lockedIDs = append(lockedIDs, id) + } + err = locked.Err() + _ = locked.Close() + if err != nil { + return 0, err + } + for _, id := range lockedIDs { + if err = accounttimer.CancelPending(ctx, exec, id, time.Now().UTC()); err != nil { + return 0, err + } + } + } result, err := exec.ExecContext(ctx, query, args...) if err != nil { return 0, err diff --git a/backend/internal/repository/account_schedulable_timer.go b/backend/internal/repository/account_schedulable_timer.go new file mode 100644 index 000000000000..241fefa9a499 --- /dev/null +++ b/backend/internal/repository/account_schedulable_timer.go @@ -0,0 +1,25 @@ +package repository + +import ( + "context" + "database/sql" + + "github.com/Wei-Shaw/sub2api/internal/accounttimer" + "github.com/Wei-Shaw/sub2api/internal/service" +) + +// SchedulableTimers exposes an optional capability without expanding the broad +// AccountRepository interface (and every unrelated mock). Production uses sql.DB. +func (r *accountRepository) SchedulableTimers() accounttimer.Repository { + db, ok := r.sql.(*sql.DB) + if !ok || db == nil { + return nil + } + return &accounttimer.Store{ + DB: db, Policy: accounttimer.ConfiguredPolicy(), + Enqueue: func(ctx context.Context, tx *sql.Tx, id int64) error { + return enqueueSchedulerOutbox(ctx, tx, service.SchedulerOutboxEventAccountChanged, &id, nil, nil) + }, + AfterCommit: r.syncSchedulerAccountSnapshotDetached, + } +} diff --git a/backend/internal/server/routes/admin.go b/backend/internal/server/routes/admin.go index 1924639921a5..6fad1c93a18e 100644 --- a/backend/internal/server/routes/admin.go +++ b/backend/internal/server/routes/admin.go @@ -474,6 +474,10 @@ func registerAccountRoutes(admin *gin.RouterGroup, h *handler.Handlers, stepUpAu accounts.GET("/:id/temp-unschedulable", h.Admin.Account.GetTempUnschedulable) accounts.DELETE("/:id/temp-unschedulable", h.Admin.Account.ClearTempUnschedulable) accounts.POST("/:id/schedulable", h.Admin.Account.SetSchedulable) + accounts.GET("/:id/schedulable-timers", h.Admin.Account.ListSchedulableTimers) + accounts.POST("/:id/schedulable-timers/preview", h.Admin.Account.PreviewSchedulableTimer) + accounts.POST("/:id/schedulable-timers", h.Admin.Account.SaveSchedulableTimer) + accounts.DELETE("/:id/schedulable-timers/:action", h.Admin.Account.CancelSchedulableTimer) accounts.POST("/models/sync-upstream-preview", h.Admin.Account.SyncUpstreamModelsPreview) accounts.GET("/:id/models", h.Admin.Account.GetAvailableModels) accounts.GET("/:id/models/reasoning", h.Admin.Account.GetModelReasoning) diff --git a/backend/internal/service/account_expiry_service.go b/backend/internal/service/account_expiry_service.go index eaada11c697b..1312047d74a8 100644 --- a/backend/internal/service/account_expiry_service.go +++ b/backend/internal/service/account_expiry_service.go @@ -5,6 +5,8 @@ import ( "log" "sync" "time" + + "github.com/Wei-Shaw/sub2api/internal/accounttimer" ) // AccountExpiryService periodically pauses expired accounts when auto-pause is enabled. @@ -33,10 +35,15 @@ func (s *AccountExpiryService) Start() { defer s.wg.Done() ticker := time.NewTicker(s.interval) defer ticker.Stop() + timerTicker := time.NewTicker(accounttimer.PollInterval) + defer timerTicker.Stop() + s.runTimers() s.runOnce() for { select { + case <-timerTicker.C: + s.runTimers() case <-ticker.C: s.runOnce() case <-s.stopCh: @@ -69,3 +76,23 @@ func (s *AccountExpiryService) runOnce() { log.Printf("[AccountExpiry] Auto paused %d expired accounts", updated) } } + +// Run the one-shot worker independently of expiry errors. Lifecycle and +// cancellation are owned by the existing background service, including Stop. +func (s *AccountExpiryService) runTimers() { + provider, ok := s.accountRepo.(interface { + SchedulableTimers() accounttimer.Repository + }) + if !ok { + return + } + timers := provider.SchedulableTimers() + if timers == nil { + return + } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := timers.RunDue(ctx); err != nil { + log.Printf("[AccountTimer] batch failed; pending rows will be reconsidered on next poll") + } +} diff --git a/backend/internal/service/account_expiry_timer_test.go b/backend/internal/service/account_expiry_timer_test.go new file mode 100644 index 000000000000..38f2152982ee --- /dev/null +++ b/backend/internal/service/account_expiry_timer_test.go @@ -0,0 +1,78 @@ +//go:build unit + +package service + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/Wei-Shaw/sub2api/internal/accounttimer" +) + +type expiryTimerRepo struct { + AccountRepository + timers accounttimer.Repository + expiry chan struct{} +} + +func (r *expiryTimerRepo) SchedulableTimers() accounttimer.Repository { return r.timers } +func (r *expiryTimerRepo) AutoPauseExpiredAccounts(context.Context, time.Time) (int64, error) { + select { + case r.expiry <- struct{}{}: + default: + } + return 0, nil +} + +type expiryTimerWorker struct { + accounttimer.Repository + ran chan struct{} +} + +func (w *expiryTimerWorker) RunDue(ctx context.Context) error { + if _, ok := ctx.Deadline(); !ok { + panic("worker context must be bounded") + } + select { + case w.ran <- struct{}{}: + default: + } + return errors.New("isolated fixture worker failure") +} +func TestAccountExpiryTimerSharesLifecycleAndDoesNotBlockExpiryOnError(t *testing.T) { + worker := &expiryTimerWorker{ran: make(chan struct{}, 2)} + repo := &expiryTimerRepo{timers: worker, expiry: make(chan struct{}, 2)} + svc := NewAccountExpiryService(repo, time.Hour) + svc.Start() + defer svc.Stop() + select { + case <-worker.ran: + case <-time.After(time.Second): + t.Fatal("timer worker not run on startup") + } + select { + case <-repo.expiry: + case <-time.After(time.Second): + t.Fatal("timer failure blocked expiry") + } + stopped := make(chan struct{}) + go func() { svc.Stop(); close(stopped) }() + select { + case <-stopped: + case <-time.After(time.Second): + t.Fatal("Stop did not join worker") + } +} +func TestAccountExpiryTimerOptionalProvider(t *testing.T) { + s := &adminServiceImpl{accountRepo: &expiryTimerRepo{}} + if s.SchedulableTimers() != nil { + t.Fatal("unexpected provider") + } + worker := &expiryTimerWorker{ran: make(chan struct{}, 1)} + s.accountRepo = &expiryTimerRepo{timers: worker} + if s.SchedulableTimers() != worker { + t.Fatal("timer capability not forwarded") + } +} diff --git a/backend/internal/service/account_schedulable_timer.go b/backend/internal/service/account_schedulable_timer.go new file mode 100644 index 000000000000..33e4f98ad89c --- /dev/null +++ b/backend/internal/service/account_schedulable_timer.go @@ -0,0 +1,14 @@ +package service + +import "github.com/Wei-Shaw/sub2api/internal/accounttimer" + +// SchedulableTimers is supplied by the production account repository. It keeps +// the timer storage separate from model tests, cooldown and quality policies. +func (s *adminServiceImpl) SchedulableTimers() accounttimer.Repository { + if provider, ok := s.accountRepo.(interface { + SchedulableTimers() accounttimer.Repository + }); ok { + return provider.SchedulableTimers() + } + return nil +} diff --git a/backend/internal/service/admin_account.go b/backend/internal/service/admin_account.go index e463ff56e695..bd70c0ffce60 100644 --- a/backend/internal/service/admin_account.go +++ b/backend/internal/service/admin_account.go @@ -15,6 +15,7 @@ import ( "strings" "time" + "github.com/Wei-Shaw/sub2api/internal/accounttimer" "github.com/Wei-Shaw/sub2api/internal/config" infraerrors "github.com/Wei-Shaw/sub2api/internal/pkg/errors" "github.com/Wei-Shaw/sub2api/internal/pkg/logger" @@ -635,6 +636,11 @@ func (s *adminServiceImpl) CreateAccount(ctx context.Context, input *CreateAccou } func (s *adminServiceImpl) UpdateAccount(ctx context.Context, id int64, input *UpdateAccountInput) (*Account, error) { + if input.Schedulable == nil { + ctx = accounttimer.WithPreservedSchedulable(ctx) + } else { + ctx = accounttimer.WithManualAccount(ctx, id) + } if err := ValidateAccountCostMultiplierExtra(input.Extra); err != nil { return nil, err } @@ -1363,6 +1369,8 @@ func (s *adminServiceImpl) BulkUpdateAccounts(ctx context.Context, input *BulkUp return nil, err } + ctx = accounttimer.WithoutManualIntent(ctx) + // 将 proxy 变更传播到每个目标账号的 spark 影子账号 if repoUpdates.ProxyID != nil { var effectiveProxyID *int64 diff --git a/backend/migrations/295_account_schedulable_timers.sql b/backend/migrations/295_account_schedulable_timers.sql new file mode 100644 index 000000000000..9d354737ebe9 --- /dev/null +++ b/backend/migrations/295_account_schedulable_timers.sql @@ -0,0 +1,35 @@ +-- One-shot admin scheduling. Independent of quality/cooldown jobs. Forward only. +-- No credentials or account/user financial state are stored or modified here. +CREATE TABLE IF NOT EXISTS account_schedulable_timers ( + account_id BIGINT NOT NULL, + action TEXT NOT NULL CHECK (action IN ('open', 'close')), + version BIGINT NOT NULL CHECK (version > 0), + execute_at TIMESTAMPTZ NOT NULL, + timezone TEXT NOT NULL, + requested_by BIGINT NOT NULL, + status TEXT NOT NULL CHECK (status IN ('pending','executed','skipped','cancelled','failed')), + attempts INTEGER NOT NULL DEFAULT 0 CHECK (attempts BETWEEN 0 AND 3), + next_attempt_at TIMESTAMPTZ NOT NULL, + actual_at TIMESTAMPTZ, + reason TEXT NOT NULL DEFAULT '', + PRIMARY KEY (account_id, action) +); +CREATE INDEX IF NOT EXISTS account_schedulable_timers_due_idx + ON account_schedulable_timers (next_attempt_at, account_id) WHERE status = 'pending'; +CREATE UNIQUE INDEX IF NOT EXISTS account_schedulable_timers_conflict_idx + ON account_schedulable_timers (account_id, execute_at) WHERE status = 'pending'; +-- Intentionally no cascading account/user FK: execution evidence survives deletion. +CREATE TABLE IF NOT EXISTS account_schedulable_timer_audit ( + id BIGSERIAL PRIMARY KEY, + account_id BIGINT NOT NULL, + action TEXT NOT NULL, + version BIGINT NOT NULL, + actor_id BIGINT NOT NULL, + execute_at TIMESTAMPTZ NOT NULL, + timezone TEXT NOT NULL, + actual_at TIMESTAMPTZ NOT NULL, + result TEXT NOT NULL, + reason TEXT NOT NULL DEFAULT '' +); +CREATE INDEX IF NOT EXISTS account_schedulable_timer_audit_account_idx + ON account_schedulable_timer_audit (account_id, id DESC); diff --git a/docs/account-schedulable-timers.md b/docs/account-schedulable-timers.md new file mode 100644 index 000000000000..b3a9f42dde67 --- /dev/null +++ b/docs/account-schedulable-timers.md @@ -0,0 +1,58 @@ +# 账号允许调度的一次性计划 + +账号管理的「调度」列，在现有立即开关上方提供「定时打开」「定时关闭」。桌面表格和移动卡片使用同一组件。每个账号可以同时保留一个打开、一个关闭计划，形成一次开关时段；不会按天重复。同动作保存会替换旧计划，相反动作不能安排在同一 UTC 瞬间。 + +## 设置时间 + +- 相对时间：正整数小时和分钟，可直接选择「8 小时后」。从保存事务获得账号锁、服务端接受计划时起算，浏览器关闭不会影响执行。预览到保存之间的等待不计入这段时长，保存后的待执行状态展示最终时刻。 +- 固定时间：日期 + `HH:mm` + IANA 时区。日期留空只预览该时区下一次这个时刻的具体日期；勾选确认后保存该明确日期及预览的 UTC 瞬间。明确填写的过期日期直接拒绝，不自动顺延。 +- 时区明确默认 `Asia/Shanghai`，有北京时间快捷入口和 UTC 快捷入口。当前账号页没有独立的用户显示时区偏好；本功能不取浏览器时区、请求时区或服务器 `TZ` 充当显示偏好。 +- 预览展示当地日期/时间、原始 IANA 时区和最终 UTC 偏移。服务端载入随二进制提供的 tzdata；夏令时/政治时区变更造成的不存在或重复本地时间均拒绝，需选择其他明确时间。 +- 最长默认 366 天，足够覆盖一年内一次性维护，避免多年遗忘的开启计划。可在所有实例统一设置 `ACCOUNT_SCHEDULABLE_TIMER_MAX_DAYS=1..3660`；无效值回退 366 天。上限十年控制误输入及时间计算范围。降低上限只影响新建/编辑，不删除已保存计划。 + +待执行条目展示动作、日期、时间、时区、偏移和相对倒计时，可以编辑或取消。倒计时以服务端返回的时间和浏览器单调时钟估计，仅用于显示。服务端每 30 秒轮询，负载、锁竞争、通知传播会增加延迟，不保证秒级准点。界面每 30 秒刷新计划状态；发现执行完成后重新读取账号列表，立即开关以服务端当前值为准。 + +## 恢复、安全和失败 + +关闭计划可在离线恢复后补执行；打开最多允许逾期 15 分钟，超过即标记 `skipped/open_expired`，必须人工重新安排。非 active、已软/硬删除或到期账号的打开也会跳过，保留可解释结果。关闭不启用账号，删除账号的关闭跳过。 + +执行仅修改 `accounts.schedulable` 和 `updated_at`，不修改 `status`、到期日期、额度、限流、临时质量隔离、手动 IP ACL、凭据或任何余额/账本/结算/支付状态。现有调度器仍检查原硬门槛，显示「打开成功」不保证账号满足其他可用条件。自动 cooldown、到期或质量策略不会带管理员手动上下文，不会取消所有计划。 + +管理员使用原立即开关或批量立即开关时会收到明确确认警告：该账号全部待执行打开/关闭计划将一起取消。后端在同一事务内完成取消、立即更新及 outbox；即使立即值已等于原值，也会取消待执行计划。管理员通过账号 PUT/bulk-update 显式发送 `schedulable` 时同样适用。普通编辑未提供该字段时，不把早先读到的旧布尔值写回，防止覆盖刚执行的计划。 + +数据库可写时，账号更新或 outbox/终态审计失败会整体回滚，按 30、60 秒退避后重试，总共最多三次执行尝试，第三次失败进入 `failed/execution_error`（终态中保留计算的 120 秒下一尝试时间，但不再被领取）。如果连数据库都不可达，无法可靠地记账尝试次数，数据库恢复后重新检查持久化待执行行，打开仍受 15 分钟补执行窗口约束。API 事务有 5 秒上下文限时；后台每批最多 50 项、5 秒上下文，避免无限阻塞。提交响应不明确时不盲目重放，下一轮重新检查状态。 + +审计表记录账号 ID、计划版本、动作、原计划操作者、到期 UTC 时刻、原始时区、实际记录时刻、结果及固定原因码。取消记录取消操作者；重新保存保留旧审计，新版本记录新的操作者。API 返回最近 20 条记录；审计不含凭据，数据库错误不会原文返回给前端。 + +## 事务与多实例 + +`account_schedulable_timers` 以 `(account_id, action)` 为主键，最多两行当前状态；终态保留版本，重新保存递增版本。pending 部分唯一索引拒绝同账号同瞬间冲突；`(next_attempt_at, account_id) WHERE status='pending'` 支持有界到期查询，不扫描账号全表。 + +所有计划保存/取消、显式手动开关、定时执行都先锁账号行，再锁计划行，遵守同一顺序。worker 使用账号 `FOR UPDATE SKIP LOCKED` 作为事务内 claim，重新核对版本、状态、重试时间和到期时刻。较晚的同账号动作等待较早 pending 动作进入终态，避免多个实例调换开关次序。取消/编辑与执行并发时，旧候选版本不能再写回；若执行已提交，取消返回版本冲突，编辑需确认最新状态。 + +账号 bool、任务终态、执行审计和现有 scheduler outbox 在同一事务提交。提交后复用账号快照同步；即使直接缓存写入失败，持久化 outbox 继续承担既有传播路径。该保证是数据库中至多一次逻辑效果，并不宣称通知只投递一次。普通修改/批量修改的取消沿用原 Ent 事务和原 outbox。到期 worker 复用 `AccountExpiryService` 的启动/停止及现有后台服务角色控制，没有 OS cron、任意 shell、SSH 或额外常驻生命周期。 + +## API（均要求管理员） + +| 方法 | 路径 | 用途 | +| --- | --- | --- | +| GET | `/api/v1/admin/accounts/:id/schedulable-timers` | 当前两项状态、20 条审计、服务端时刻、轮询/补执行/时长限制 | +| POST | `/api/v1/admin/accounts/:id/schedulable-timers/preview` | 解析时间并返回 UTC 和当地预览 | +| POST | `/api/v1/admin/accounts/:id/schedulable-timers` | 保存，包含 `action` 和 `expected_version` | +| DELETE | `/api/v1/admin/accounts/:id/schedulable-timers/:action?version=N` | 版本匹配时取消 | + +保存相对时间例：`{"mode":"relative","hours":8,"minutes":0,"timezone":"Asia/Shanghai","action":"open","expected_version":0}`。固定保存要求明确 `date`、`time`、`timezone`、`confirmed_at`（与服务端重新解析的 UTC 瞬间一致）。首次版本 0，后续从 GET 读取，旧版本返回 409；相反动作同瞬间返回 409。观察员与普通用户不能读写新定时 API；原管理路由范围/认证机制继续生效。 + +## 迁移与回退 + +新增 `295_account_schedulable_timers.sql`，遵守不可变、forward-only 迁移规范，只增加两张表及索引，不修改现有账号 schema。没有 Ent schema 变更或 Wire 依赖生成要求。 + +回退到旧二进制时，旧代码可以忽略附加表，但不会执行、展示或取消这些计划。因此不能直接回退后又无审查地重新升级：先由新版本管理员取消 pending 计划并停止新版本 worker，留存审计，再回退；旧二进制期间创建的手动意图无法自动记录到新计划表。若将来删除表，应单独编写后续迁移，保留所需审计，不向本文件追加 Down SQL。本轮只验证最小隔离 fixture 的增删表兼容性和静态迁移约束，未运行应用迁移入口或真实迁移文件。 + +## 定向验证 + +核心包只有 stdlib/SQL 依赖，管理员 timer HTTP 子包独立于大型 service 编译。使用 Go 1.27.3，`GOMAXPROCS=1 GOMEMLIMIT=600MiB`，数据盘 `GOCACHE/GOTMPDIR/GOPATH`，`go test -p 1 ./internal/accounttimer ./internal/handler/admin/accounttimerapi`。 + +集成测试 `-tags integration ./internal/accounttimer` 需要显式 `ACCOUNT_TIMER_TEST_DSN`，只接受 loopback 下的 `schedulable_timer_fixture` 测试数据库。每个测试自行新建唯一 schema 和最小 fixtures，结束后删除自己的 schema；不使用应用数据、真实模型或迁移 runner。测试覆盖模拟时钟、时区/DST、同瞬间冲突、版本替换/取消、启动恢复、多 worker、有限失败重试、账号删除/到期/禁用、通知回滚和硬门槛哨兵。 + +前端定向 Vitest 位于 `SchedulableTimerControl.spec.ts`，另核对中英键完整性和现有账号页测试。桌面/移动浏览器截图来自真实 DataTable + 新组件的隔离 mock harness，完整账号页和全栈真实缓存传播仍需后续 CI/集成复核。 diff --git a/frontend/src/api/admin/schedulableTimers.ts b/frontend/src/api/admin/schedulableTimers.ts new file mode 100644 index 000000000000..e9ffeb8ed03b --- /dev/null +++ b/frontend/src/api/admin/schedulableTimers.ts @@ -0,0 +1,66 @@ +import { apiClient } from '../client' + +export type TimerAction = 'open' | 'close' +export interface TimerInput { + mode: 'relative' | 'absolute' + hours: number + minutes: number + date: string + time: string + timezone: string +} +export interface TimerPreview { + execute_at: string + timezone: string + local_date: string + local_time: string + offset: string + server_now: string +} +export interface SchedulableTimer { + account_id: number + action: TimerAction + version: number + execute_at: string + timezone: string + local_date?: string + local_time?: string + offset?: string + requested_by: number + status: 'pending' | 'executed' | 'skipped' | 'cancelled' | 'failed' + attempts: number + actual_at: string | null + reason: string +} +export interface TimerAudit { + action: TimerAction + version: number + actor_id: number + execute_at: string + timezone: string + actual_at: string + result: string + reason: string +} +export interface TimerSnapshot { + tasks: SchedulableTimer[] + history: TimerAudit[] + server_now: string + poll_seconds: number + open_grace_seconds: number + max_delay_minutes: number +} +export const schedulableTimersAPI = { + async list(id: number) { + return (await apiClient.get(`/admin/accounts/${id}/schedulable-timers`)).data + }, + async preview(id: number, input: TimerInput) { + return (await apiClient.post(`/admin/accounts/${id}/schedulable-timers/preview`, input)).data + }, + async save(id: number, input: TimerInput & { action: TimerAction; expected_version: number; confirmed_at?: string }) { + return (await apiClient.post(`/admin/accounts/${id}/schedulable-timers`, input)).data + }, + async cancel(id: number, action: TimerAction, version: number) { + await apiClient.delete(`/admin/accounts/${id}/schedulable-timers/${action}`, { params: { version } }) + } +} diff --git a/frontend/src/components/account/SchedulableTimerControl.vue b/frontend/src/components/account/SchedulableTimerControl.vue new file mode 100644 index 000000000000..e4fc812b7cb0 --- /dev/null +++ b/frontend/src/components/account/SchedulableTimerControl.vue @@ -0,0 +1,202 @@ + + + diff --git a/frontend/src/components/account/__tests__/SchedulableTimerControl.spec.ts b/frontend/src/components/account/__tests__/SchedulableTimerControl.spec.ts new file mode 100644 index 000000000000..fb5cc0d779d1 --- /dev/null +++ b/frontend/src/components/account/__tests__/SchedulableTimerControl.spec.ts @@ -0,0 +1,109 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { flushPromises, mount, type VueWrapper } from '@vue/test-utils' +import SchedulableTimerControl from '../SchedulableTimerControl.vue' + +const { api, toast } = vi.hoisted(() => ({ + api: { list: vi.fn(), preview: vi.fn(), save: vi.fn(), cancel: vi.fn() }, + toast: { showSuccess: vi.fn(), showError: vi.fn() } +})) +vi.mock('@/api/admin/schedulableTimers', () => ({ schedulableTimersAPI: api })) +vi.mock('@/stores/app', () => ({ useAppStore: () => toast })) +vi.mock('vue-i18n', () => ({ useI18n: () => ({ t: (key: string) => key }) })) +const task = { account_id: 7, action: 'open', version: 3, execute_at: '2026-10-23T00:00:00Z', timezone: 'Asia/Shanghai', requested_by: 9, status: 'pending', attempts: 0, actual_at: null, reason: '' } +const snapshot = () => ({ tasks: [], history: [], server_now: '2026-10-22T00:00:00Z', poll_seconds: 30, open_grace_seconds: 900, max_delay_minutes: 527040 }) +let wrapper: VueWrapper +beforeEach(() => { + vi.clearAllMocks() + api.list.mockResolvedValue(snapshot()) + api.preview.mockResolvedValue({ execute_at: task.execute_at, timezone: 'Asia/Shanghai', local_date: '2026-10-23', local_time: '08:00', offset: '+08:00', server_now: '2026-10-22T00:00:00Z' }) + api.save.mockResolvedValue(task) + api.cancel.mockResolvedValue(undefined) +}) +afterEach(() => wrapper?.unmount()) +describe('one-shot schedulable timer', () => { + it('expands compact controls and saves 8h using server relative input', async () => { + wrapper = mount(SchedulableTimerControl, { props: { accountId: 7 } }) + await flushPromises() + expect(wrapper.find('[data-test="timer-form"]').exists()).toBe(false) + await wrapper.get('[data-test="timer-open"]').trigger('click') + await flushPromises() + expect(wrapper.get('[data-test="timer-zone"]').element).toHaveProperty('value', 'Asia/Shanghai') + await wrapper.get('[data-test="timer-preview"]').trigger('click') + await flushPromises() + await wrapper.get('[data-test="timer-save"]').trigger('click') + await flushPromises() + expect(api.save).toHaveBeenCalledWith(7, expect.objectContaining({ mode: 'relative', hours: 8, minutes: 0, action: 'open', expected_version: 0 })) + expect(toast.showSuccess).toHaveBeenCalled() + }) + it('requires confirmation of the concrete local date, UTC offset and zone for HH:mm only', async () => { + wrapper = mount(SchedulableTimerControl, { props: { accountId: 7 } }) + await flushPromises() + await wrapper.get('[data-test="timer-close"]').trigger('click') + await flushPromises() + await wrapper.get('[data-test="timer-mode"]').setValue('absolute') + await wrapper.get('[data-test="timer-preview"]').trigger('click') + await flushPromises() + expect(wrapper.text()).toContain('2026-10-23') + expect(wrapper.text()).toContain('+08:00') + expect(wrapper.text()).toContain('Asia/Shanghai') + expect(wrapper.get('[data-test="timer-save"]').attributes('disabled')).toBeDefined() + await wrapper.get('[data-test="timer-confirm"]').setValue(true) + await wrapper.get('[data-test="timer-save"]').trigger('click') + await flushPromises() + expect(api.save).toHaveBeenCalledWith(7, expect.objectContaining({ action: 'close', date: '2026-10-23', confirmed_at: task.execute_at, hours: 0 })) + }) + it('reloads saved tasks, edits by version and cancels the chosen action', async () => { + api.list.mockResolvedValue({ ...snapshot(), tasks: [task] }) + wrapper = mount(SchedulableTimerControl, { props: { accountId: 7 } }) + await flushPromises() + expect(wrapper.text()).toContain('2026-10-23') + await wrapper.get('[data-test="timer-open"]').trigger('click') + await flushPromises() + expect(wrapper.get('[data-test="timer-mode"]').element).toHaveProperty('value', 'absolute') + await wrapper.get('[data-test="timer-preview"]').trigger('click') + await flushPromises() + await wrapper.get('[data-test="timer-confirm"]').setValue(true) + await wrapper.get('[data-test="timer-save"]').trigger('click') + await flushPromises() + expect(api.save).toHaveBeenCalledWith(7, expect.objectContaining({ expected_version: 3 })) + await wrapper.get('[data-test="timer-cancel-open"]').trigger('click') + await flushPromises() + expect(api.cancel).toHaveBeenCalledWith(7, 'open', 3) + }) + it('shows translated validation errors and clears obsolete previews after editing', async () => { + wrapper = mount(SchedulableTimerControl, { props: { accountId: 7 } }) + await flushPromises() + await wrapper.get('[data-test="timer-open"]').trigger('click') + await flushPromises() + await wrapper.get('[data-test="timer-preview"]').trigger('click') + await flushPromises() + await wrapper.get('[data-test="timer-zone"]').setValue('Invalid/Zone') + expect(wrapper.find('[data-test="timer-save"]').exists()).toBe(false) + api.preview.mockRejectedValue({ response: { data: { reason: 'TIMER_timezone' } } }) + await wrapper.get('[data-test="timer-preview"]').trigger('click') + await flushPromises() + expect(wrapper.text()).toContain('timer.errors.timezone') + }) + it('refreshes after manual changes and keeps an accessible narrow-screen form', async () => { + wrapper = mount(SchedulableTimerControl, { props: { accountId: 7, refreshKey: 0 } }) + await flushPromises() + await wrapper.setProps({ refreshKey: 1 }) + await flushPromises() + expect(api.list).toHaveBeenCalledTimes(2) + await wrapper.get('[data-test="timer-open"]').trigger('click') + await flushPromises() + expect(wrapper.get('[data-test="timer-open"]').attributes('aria-expanded')).toBe('true') + expect(wrapper.get('[data-test="timer-form"]').classes()).toContain('max-w-full') + expect(wrapper.get('[data-test="timer-zone"]').attributes('aria-label')).toBeTruthy() + }) + it('requests account reconciliation after a persisted execution changes the switch', async () => { + api.list.mockResolvedValueOnce({ ...snapshot(), tasks: [task] }) + wrapper = mount(SchedulableTimerControl, { props: { accountId: 7 } }) + await flushPromises() + api.list.mockResolvedValue({ ...snapshot(), tasks: [{ ...task, status: 'executed', actual_at: task.execute_at }] }) + await wrapper.setProps({ refreshKey: 1 }) + await flushPromises() + expect(wrapper.emitted('executed')).toHaveLength(1) + }) + +}) diff --git a/frontend/src/i18n/locales/en/admin/accounts.ts b/frontend/src/i18n/locales/en/admin/accounts.ts index d0985aa81040..8fc01946e212 100644 --- a/frontend/src/i18n/locales/en/admin/accounts.ts +++ b/frontend/src/i18n/locales/en/admin/accounts.ts @@ -1,5 +1,66 @@ export default { accounts: { + timer: { + "open": "Timed on", + "close": "Timed off", + "edit": "Edit", + "cancel": "Cancel", + "reload": "Retry loading", + "mode": "Time mode", + "relative": "After a duration", + "absolute": "At a local date/time", + "hours": "Hours", + "minutes": "Minutes", + "preset": "8 hours", + "relativeHint": "Duration starts when the server accepts Save. The preview is approximate until saved.", + "date": "Date (optional)", + "time": "Local time", + "absoluteHint": "Leave the date empty to preview the next occurrence. Confirm the specific date before saving. A past explicit date is rejected. DST gaps and repeated times are rejected.", + "timezone": "IANA time zone", + "beijing": "Beijing · Asia/Shanghai", + "zoneHint": "Default: Asia/Shanghai. This is explicit and does not follow browser or server time zone.", + "preview": "Preview execution time", + "confirm": "I confirm this specific date, time and time zone. One execution only.", + "save": "Save one-shot timer", + "collapse": "Collapse", + "policy": "Polls every {seconds}s; execution may be later. Opens over {grace} min late are skipped; closes run after recovery. Maximum horizon: {days} days. Scheduling changes only scheduling permission; all account safety gates still apply.", + "history": "Recent results", + "actual": "Recorded at", + "awaiting": "Due; waiting for worker", + "remaining": "in {hours}h {minutes}m", + "saved": "Timer saved", + "cancelled": "Timer cancelled", + "manualWarning": "Switch scheduling permission immediately? This also cancels ALL pending timed-on and timed-off actions for the affected account(s), even if the permission already has this value.", + "results": { + "scheduled": "Scheduled", + "replaced": "Replaced", + "cancelled": "Cancelled", + "executed": "Executed", + "skipped": "Skipped", + "failed": "Failed", + "pending": "Retry pending" + }, + "errors": { + "timezone": "Enter a valid IANA time zone (for example Asia/Shanghai).", + "datetime": "Enter a valid date and HH:mm.", + "dst": "This local time is missing or ambiguous because of a time-zone transition. Choose another time.", + "duration": "Use a positive whole number of hours/minutes within the configured horizon.", + "mode": "Choose a valid time mode.", + "past": "This date/time has passed. Choose a future date explicitly.", + "confirmation": "Preview and confirm the specific date/time before saving.", + "version": "The plan changed or already ran. Reload and review before editing again.", + "conflict": "Timed on and timed off cannot use the same instant.", + "request": "Invalid timer request.", + "account_deleted": "Account deleted or unavailable.", + "account_inactive": "Skipped: account is inactive; its status was preserved.", + "account_expired": "Skipped: account has expired.", + "open_expired": "Skipped: timed on exceeded the 15-minute recovery window.", + "execution_error": "Database/outbox update failed. Retries are bounded to three attempts; a failed plan must be reviewed and rescheduled.", + "admin_cancelled": "Cancelled by an administrator.", + "manual_override": "Cancelled by an immediate manual scheduling change.", + "internal": "Could not complete the timer operation. Retry or refresh to check its current state." + } +}, bpsAuthorizing: 'BPS authorization is automatic. Native routing remains active until authorization succeeds, then BPS activates automatically.', "modelsLoadFailed": "Could not load models. Please retry.", "retryModels": "Reload models", diff --git a/frontend/src/i18n/locales/zh/admin/accounts.ts b/frontend/src/i18n/locales/zh/admin/accounts.ts index dedd133c1e22..5a009b9c1068 100644 --- a/frontend/src/i18n/locales/zh/admin/accounts.ts +++ b/frontend/src/i18n/locales/zh/admin/accounts.ts @@ -1,5 +1,66 @@ export default { accounts: { + timer: { + "open": "定时打开", + "close": "定时关闭", + "edit": "编辑", + "cancel": "取消计划", + "reload": "重新加载", + "mode": "时间方式", + "relative": "一段时间后", + "absolute": "指定当地日期/时间", + "hours": "小时", + "minutes": "分钟", + "preset": "8 小时后", + "relativeHint": "从服务端接受“保存”的时刻起算；保存前的预览为预计时间。", + "date": "日期（可留空）", + "time": "当地时间", + "absoluteHint": "日期留空将预览该时区下一次此时刻，保存前须确认具体日期。已过的明确日期会被拒绝；夏令时不存在或重复的时刻也会被拒绝。", + "timezone": "IANA 时区", + "beijing": "北京时间 · Asia/Shanghai", + "zoneHint": "明确默认 Asia/Shanghai，不跟随浏览器或服务器时区。", + "preview": "预览执行时间", + "confirm": "我确认此具体日期、时刻和时区，仅执行一次。", + "save": "保存一次性计划", + "collapse": "收起", + "policy": "每 {seconds} 秒轮询，实际执行可能延后。打开逾期 {grace} 分钟跳过；关闭在恢复后补执行。最长可设 {days} 天。仅修改允许调度，账号的其他安全门槛仍然生效。", + "history": "近期执行记录", + "actual": "记录时间", + "awaiting": "已到期，等待后台处理", + "remaining": "剩余 {hours} 小时 {minutes} 分钟", + "saved": "定时计划已保存", + "cancelled": "定时计划已取消", + "manualWarning": "立即切换允许调度？此操作会同时取消所影响账号的全部待执行“定时打开”和“定时关闭”计划，即使当前开关已经是此值。", + "results": { + "scheduled": "已安排", + "replaced": "已替换", + "cancelled": "已取消", + "executed": "已执行", + "skipped": "已跳过", + "failed": "执行失败", + "pending": "等待重试" + }, + "errors": { + "timezone": "请输入有效的 IANA 时区，例如 Asia/Shanghai。", + "datetime": "请输入有效的日期和 HH:mm 时刻。", + "dst": "该当地时间因时区变化不存在或有歧义，请选择其他时刻。", + "duration": "请输入正整数小时/分钟，总时长须在配置上限内。", + "mode": "请选择有效的时间方式。", + "past": "此日期时刻已经过去，请明确选择未来日期。", + "confirmation": "请先预览并确认具体执行日期和时刻。", + "version": "计划已被修改或已执行，请重新加载并核对后编辑。", + "conflict": "定时打开与定时关闭不能安排在同一瞬间。", + "request": "定时请求无效。", + "account_deleted": "账号已删除或不可用。", + "account_inactive": "已跳过：账号非启用状态，原状态保持不变。", + "account_expired": "已跳过：账号已到期。", + "open_expired": "已跳过：定时打开超过 15 分钟补执行窗口。", + "execution_error": "数据库或调度通知写入失败，最多尝试三次；失败后需检查并重新设定计划。", + "admin_cancelled": "管理员取消了此计划。", + "manual_override": "管理员立即切换调度开关，已取消此计划。", + "internal": "定时操作未完成，请重试或刷新确认当前状态。" + } +}, bpsAuthorizing: 'BPS 正在自动授权，当前使用原生通道；授权成功后自动切换，无需再次操作开关。', "modelsLoadFailed": "模型列表加载失败，请重试。", "retryModels": "重新加载模型", diff --git a/frontend/src/views/admin/AccountsView.vue b/frontend/src/views/admin/AccountsView.vue index 103fc491a18b..f2a662a39128 100644 --- a/frontend/src/views/admin/AccountsView.vue +++ b/frontend/src/views/admin/AccountsView.vue @@ -296,6 +296,7 @@