wehub-resource-sync
bf9395e022
CI / license-header (push) Has been skipped
CI / e2e-dry-run (push) Has been skipped
CI / fast-gate (push) Failing after 0s
Test PR Label Logic / test-pr-labels (push) Failing after 1s
Skill Format Check / check-format (push) Failing after 2s
CI / security (push) Failing after 5s
CI / unit-test (push) Has been skipped
CI / lint (push) Has been skipped
CI / script-test (push) Has been skipped
CI / deterministic-gate (push) Has been skipped
CI / coverage (push) Has been skipped
CI / results (push) Has been cancelled
CI / deadcode (push) Has been cancelled
CI / e2e-live (push) Has been cancelled
293 lines
11 KiB
Go
293 lines
11 KiB
Go
// 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 <app_id> --target 2h",
|
||
"Apply with +db-recovery-apply --target <same> --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 <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, ", ")
|
||
}
|