// Copyright (c) 2026 Lark Technologies Pte. Ltd. // SPDX-License-Identifier: MIT package apps import ( "context" "fmt" "io" "strings" "time" "github.com/larksuite/cli/errs" "github.com/larksuite/cli/shortcuts/common" ) const dbRecoveryHint = "PITR window is up to 7 days back, limited by your last `+db-env-migrate`; pass --target as a time (e.g. 2h / 2026-04-15 / 2026-04-15T10:00:00Z)" // AppsDBRecoveryDiff 预览把数据库恢复到某个时间点会带来的变更(PITR diff,不落地)。 // // POST /apps/{app_id}/db/env_recovery,body {target, dry_run:true} → preview_request_id, // 轮询 env_recovery_diff_status 至终态,返回受影响表与行数变化。预览也需 spark:app:write scope。 var AppsDBRecoveryDiff = common.Shortcut{ Service: appsService, Command: "+db-recovery-diff", Description: "Preview restoring the database to a point in time (PITR diff)", Risk: "read", Tips: []string{ "Example: lark-cli apps +db-recovery-diff --app-id --target 2h", "Apply with +db-recovery-apply --target --yes.", }, Scopes: []string{"spark:app:write"}, AuthTypes: []string{"user"}, HasFormat: true, Flags: append([]common.Flag{ {Name: "app-id", Desc: "Miaoda app id", Required: true}, {Name: "target", Desc: "point in time to restore to; relative (2h/3d) | date | datetime | ISO 8601 w/ TZ", Required: true}, }, dbEnvFlags("", []string{"dev", "online"}, "target db environment; leave unset to auto-select (multi-env app uses dev, single-env uses online), or pass dev/online")...), Validate: func(ctx context.Context, rctx *common.RuntimeContext) error { if _, err := requireAppID(rctx.Str("app-id")); err != nil { return err } if err := rejectLegacyEnvFlag(rctx); err != nil { return err } return normalizeTimeFlags(rctx, "target") }, DryRun: func(ctx context.Context, rctx *common.RuntimeContext) *common.DryRunAPI { appID, _ := requireAppID(rctx.Str("app-id")) return common.NewDryRunAPI().POST(appRecoveryPath(appID)).Desc("Preview PITR recovery"). Params(dbEnvParams(rctx, map[string]interface{}{})). Body(map[string]interface{}{"target": rctx.Str("target"), "dry_run": true}) }, Execute: func(ctx context.Context, rctx *common.RuntimeContext) error { appID, err := requireAppID(rctx.Str("app-id")) if err != nil { return err } target := rctx.Str("target") preview, err := runRecoveryPreview(rctx, appID, target) if err != nil { return err } out := recoveryDiffOutput(target, preview) rctx.OutFormat(out, nil, func(w io.Writer) { renderRecoveryDiff(w, target, out) }) return nil }, } // AppsDBRecoveryApply 把数据库恢复到某个时间点(覆盖当前数据,异步,CLI 轮询至完成)。 // // POST /apps/{app_id}/db/env_recovery,body {target, dry_run:false};目标=当前态时短路 no_changes, // 否则轮询 env_recovery_apply_status 至 success。high-risk-write。 var AppsDBRecoveryApply = common.Shortcut{ Service: appsService, Command: "+db-recovery-apply", Description: "Restore the database to a point in time (overwrites current data, irreversible)", Risk: "high-risk-write", Tips: []string{ "Example: lark-cli apps +db-recovery-apply --app-id --target 2026-04-15T10:00:00Z --yes", "Preview first with +db-recovery-diff.", }, Scopes: []string{"spark:app:write"}, AuthTypes: []string{"user"}, HasFormat: true, Flags: append([]common.Flag{ {Name: "app-id", Desc: "Miaoda app id", Required: true}, {Name: "target", Desc: "point in time to restore to; relative (2h/3d) | date | datetime | ISO 8601 w/ TZ", Required: true}, }, dbEnvFlags("", []string{"dev", "online"}, "target db environment; leave unset to auto-select (multi-env app uses dev, single-env uses online), or pass dev/online")...), Validate: func(ctx context.Context, rctx *common.RuntimeContext) error { if _, err := requireAppID(rctx.Str("app-id")); err != nil { return err } if err := rejectLegacyEnvFlag(rctx); err != nil { return err } return normalizeTimeFlags(rctx, "target") }, DryRun: func(ctx context.Context, rctx *common.RuntimeContext) *common.DryRunAPI { appID, _ := requireAppID(rctx.Str("app-id")) return common.NewDryRunAPI().POST(appRecoveryPath(appID)).Desc("Apply PITR recovery"). Params(dbEnvParams(rctx, map[string]interface{}{})). Body(map[string]interface{}{"target": rctx.Str("target"), "dry_run": false}) }, Execute: func(ctx context.Context, rctx *common.RuntimeContext) error { appID, err := requireAppID(rctx.Str("app-id")) if err != nil { return err } target := rctx.Str("target") stop := rctx.StartSpinner("Restoring database (target: " + target + ")") defer stop() submit, err := rctx.CallAPITyped("POST", appRecoveryPath(appID), dbEnvParams(rctx, map[string]interface{}{}), map[string]interface{}{"target": target, "dry_run": false}) if err != nil { return withAppsHint(err, dbRecoveryHint) } // 目标=当前态 → 后端短路 no_changes,不轮询。 if strings.ToLower(common.GetString(submit, "status")) == "no_changes" { stop() out := map[string]interface{}{"status": "no_changes", "target": target} rctx.OutFormat(out, nil, func(w io.Writer) { io.WriteString(w, "No changes — database is already at this state.\n") }) return nil } final, perr := pollUntil(rctx.Ctx(), 2*time.Second, 2*time.Minute, func() (map[string]interface{}, error) { return rctx.CallAPITyped("GET", appRecoveryApplyStatusPath(appID), dbEnvParams(rctx, map[string]interface{}{}), nil) }, func(d map[string]interface{}) (bool, error) { switch strings.ToLower(common.GetString(d, "status")) { case "success", "restored", "ready": return true, nil case "failed": msg := common.GetString(d, "error_message") if msg == "" { msg = fmt.Sprintf("recovery to %s failed", target) } return false, withAppsHint(errs.NewAPIError(errs.SubtypeServerError, "%s", msg), dbRecoveryHint) } return false, nil }) if perr != nil { return perr } stop() out := map[string]interface{}{"status": "restored", "target": target} if n := intFromAny(final["restore_time_sec"]); n > 0 { out["restore_time_sec"] = n } rctx.OutFormat(out, nil, func(w io.Writer) { if n, ok := out["restore_time_sec"].(int); ok { fmt.Fprintf(w, "✓ Database restored to %s (%ds elapsed)\n", target, n) } else { fmt.Fprintf(w, "✓ Database restored to %s\n", target) } }) return nil }, } // runRecoveryPreview 触发 PITR 预览(dry_run=true)拿 preview_request_id,轮询 diff_status 至终态。 func runRecoveryPreview(rctx *common.RuntimeContext, appID, target string) (map[string]interface{}, error) { stop := rctx.StartSpinner("Previewing recovery impact (target: " + target + ")") defer stop() submit, err := rctx.CallAPITyped("POST", appRecoveryPath(appID), dbEnvParams(rctx, map[string]interface{}{}), map[string]interface{}{"target": target, "dry_run": true}) if err != nil { return nil, withAppsHint(err, dbRecoveryHint) } prid := common.GetString(submit, "preview_request_id") if prid == "" { return nil, errs.NewInternalError(errs.SubtypeInvalidResponse, "recovery diff did not return preview_request_id") } return pollUntil(rctx.Ctx(), 1*time.Second, 2*time.Minute, func() (map[string]interface{}, error) { return rctx.CallAPITyped("GET", appRecoveryDiffStatusPath(appID), dbEnvParams(rctx, map[string]interface{}{"preview_request_id": prid}), nil) }, func(d map[string]interface{}) (bool, error) { switch strings.ToLower(common.GetString(d, "preview_status")) { case "success": return true, nil case "failed": msg := common.GetString(d, "error_message") if msg == "" { msg = "recovery preview failed" } return false, withAppsHint(errs.NewAPIError(errs.SubtypeServerError, "%s", msg), dbRecoveryHint) } return false, nil }) } type recoveryChange struct { Table string `json:"table"` Inserted interface{} `json:"inserted,omitempty"` Deleted interface{} `json:"deleted,omitempty"` Action string `json:"action,omitempty"` DroppedAt string `json:"dropped_at,omitempty"` } // recoveryDiffOutput 组装 diff 输出:target / tables_affected / changes[] / estimated_seconds。 func recoveryDiffOutput(target string, preview map[string]interface{}) map[string]interface{} { arr, _ := preview["changes"].([]interface{}) raw := make([]recoveryChange, 0, len(arr)) for _, it := range arr { m, ok := it.(map[string]interface{}) if !ok { continue } raw = append(raw, recoveryChange{ Table: common.GetString(m, "table"), Inserted: m["inserted"], Deleted: m["deleted"], Action: common.GetString(m, "action"), DroppedAt: common.GetString(m, "dropped_at"), }) } // 服务端可能对同一张表既下发 schema 动作(drop/restore/alter)、又下发纯数据行变更。 // schema 动作已涵盖数据结果(如 drop 隐含删光行),丢弃该表的冗余数据行那条,避免同表 // 两行 + tables_affected 翻倍。 hasSchema := map[string]bool{} for _, c := range raw { if c.Action != "" { hasSchema[c.Table] = true } } changes := make([]recoveryChange, 0, len(raw)) for _, c := range raw { if c.Action == "" && hasSchema[c.Table] { continue } changes = append(changes, c) } // tables_affected 按去重后的不同表数计(而非变更条数)。 seen := map[string]bool{} for _, c := range changes { seen[c.Table] = true } est := intFromAny(preview["estimated_seconds"]) if est == 0 { est = 30 // PRD 兜底 } return map[string]interface{}{ "target": target, "tables_affected": len(seen), "changes": changes, "estimated_seconds": est, } } // renderRecoveryDiff 渲染 PITR 恢复预览:受影响表数、逐表变化描述及预估耗时;无变更打提示。 func renderRecoveryDiff(w io.Writer, target string, out map[string]interface{}) { changes, _ := out["changes"].([]recoveryChange) if len(changes) == 0 { io.WriteString(w, "No changes — database is already at this state.\n") return } fmt.Fprintf(w, "Recovery preview (→ %s):\n\n", target) fmt.Fprintf(w, " tables affected: %d\n", intFromAny(out["tables_affected"])) for _, c := range changes { fmt.Fprintf(w, " %s: %s\n", c.Table, describeRecoveryChange(c)) } fmt.Fprintf(w, "\n estimated time: ~%ds\n", intFromAny(out["estimated_seconds"])) } // describeRecoveryChange:schema 动作 或 数据行变化二选一(无 modified,对齐设计)。 func describeRecoveryChange(c recoveryChange) string { switch c.Action { case "restore_table": return "table will be restored" case "drop_table": return "table will be dropped" case "alter_table": return "table will be altered" case "unavailable": if c.DroppedAt != "" { return "diff unavailable: " + c.DroppedAt } return "diff unavailable" } parts := make([]string, 0, 2) if n := intFromAny(c.Inserted); n != 0 { parts = append(parts, fmt.Sprintf("+%d rows", n)) } if n := intFromAny(c.Deleted); n != 0 { parts = append(parts, fmt.Sprintf("-%d rows", n)) } if len(parts) == 0 { return "no changes" } return strings.Join(parts, ", ") }