| 1 | package catalog |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "database/sql" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "os" |
| 9 | "path/filepath" |
| 10 | "time" |
| 11 | ) |
| 12 | |
| 13 | // Delete permanently removes a disabled tenant. The deleting state and intent |
| 14 | // make the operation safe to resume after a process or host failure. |
| 15 | func (c *Catalog) Delete(ctx context.Context, id string) error { |
| 16 | if err := contextError(ctx); err != nil { |
| 17 | return err |
| 18 | } |
| 19 | root := c.tenantPaths(id).Root |
| 20 | staging := filepath.Join(filepath.Dir(root), ".deleting-" +id) |
| 21 | if err := c.beginDeletion(ctx, id, root, staging); err != nil { |
| 22 | return err |
| 23 | } |
| 24 | if err := stageTenant(root, staging); err != nil { |
| 25 | return err |
| 26 | } |
| 27 | if err := os.RemoveAll(staging); err != nil { |
| 28 | return fmt.Errorf("remove tenant data: %w" , err) |
| 29 | } |
| 30 | return c.finishDeletion(ctx, id) |
| 31 | } |
| 32 | |
| 33 | func (c *Catalog) beginDeletion(ctx context.Context, id, root, staging string) error { |
| 34 | return c.dbTransaction(ctx, func(tx *sql.Tx) error { |
| 35 | var status Status |
| 36 | if err := tx.QueryRowContext(ctx, "SELECT status FROM tenants WHERE id=?" , id).Scan(&status); errors.Is(err, sql.ErrNoRows) { |
| 37 | return ErrNotFound |
| 38 | } else if err != nil { |
| 39 | return err |
| 40 | } else if status != StatusDisabled { |
| 41 | return ErrInvalidTransition |
| 42 | } |
| 43 | now := time.Now().UTC().UnixNano() |
| 44 | if _, err := tx.ExecContext(ctx, "UPDATE tenants SET status=?,updated_at=? WHERE id=? AND status=?" , StatusDeleting, now, id, StatusDisabled); err != nil { |
| 45 | return fmt.Errorf("mark tenant deleting: %w" , err) |
| 46 | } |
| 47 | if _, err := tx.ExecContext(ctx, "INSERT INTO tenant_deletions(id,root,staging,started_at) VALUES(?,?,?,?)" , id, root, staging, now); err != nil { |
| 48 | return fmt.Errorf("record tenant deletion: %w" , err) |
| 49 | } |
| 50 | return nil |
| 51 | }) |
| 52 | } |
| 53 | |
| 54 | func (c *Catalog) finishDeletion(ctx context.Context, id string) error { |
| 55 | return c.dbTransaction(ctx, func(tx *sql.Tx) error { |
| 56 | if _, err := tx.ExecContext(ctx, "DELETE FROM tenant_hosts WHERE tenant_id=?" , id); err != nil { |
| 57 | return fmt.Errorf("delete tenant hosts: %w" , err) |
| 58 | } |
| 59 | result, err := tx.ExecContext(ctx, "DELETE FROM tenants WHERE id=? AND status=?" , id, StatusDeleting) |
| 60 | if err != nil { |
| 61 | return fmt.Errorf("delete tenant catalog row: %w" , err) |
| 62 | } |
| 63 | count, err := result.RowsAffected() |
| 64 | if err != nil { |
| 65 | return err |
| 66 | } |
| 67 | if count != 1 { |
| 68 | return ErrNotFound |
| 69 | } |
| 70 | _, err = tx.ExecContext(ctx, "DELETE FROM tenant_deletions WHERE id=?" , id) |
| 71 | return err |
| 72 | }) |
| 73 | } |
| 74 | |
| 75 | func (c *Catalog) recoverDeletions(ctx context.Context) error { |
| 76 | rows, err := c.db.QueryContext(ctx, "SELECT id,root,staging FROM tenant_deletions ORDER BY started_at,id" ) |
| 77 | if err != nil { |
| 78 | return fmt.Errorf("scan tenant deletions: %w" , err) |
| 79 | } |
| 80 | defer rows.Close() |
| 81 | type intent struct{ id, root, staging string } |
| 82 | var intents []intent |
| 83 | for rows.Next() { |
| 84 | var item intent |
| 85 | if err := rows.Scan(&item.id, &item.root, &item.staging); err != nil { |
| 86 | return err |
| 87 | } |
| 88 | intents = append(intents, item) |
| 89 | } |
| 90 | if err := rows.Err(); err != nil { |
| 91 | return err |
| 92 | } |
| 93 | for _, item := range intents { |
| 94 | if err := stageTenant(item.root, item.staging); err != nil { |
| 95 | return fmt.Errorf("recover tenant %q: %w" , item.id, err) |
| 96 | } |
| 97 | if err := os.RemoveAll(item.staging); err != nil { |
| 98 | return fmt.Errorf("recover tenant data %q: %w" , item.id, err) |
| 99 | } |
| 100 | if err := c.finishDeletion(ctx, item.id); err != nil { |
| 101 | return fmt.Errorf("recover tenant catalog %q: %w" , item.id, err) |
| 102 | } |
| 103 | } |
| 104 | return nil |
| 105 | } |
| 106 | |
| 107 | func stageTenant(root, staging string) error { |
| 108 | if _, err := os.Stat(root); errors.Is(err, os.ErrNotExist) { |
| 109 | if _, stagingErr := os.Stat(staging); stagingErr == nil || errors.Is(stagingErr, os.ErrNotExist) { |
| 110 | return nil |
| 111 | } else { |
| 112 | return stagingErr |
| 113 | } |
| 114 | } else if err != nil { |
| 115 | return err |
| 116 | } |
| 117 | return os.Rename(root, staging) |
| 118 | } |
| 119 | |
| 120 | func (c *Catalog) dbTransaction(ctx context.Context, fn func(*sql.Tx) error) error { |
| 121 | tx, err := c.db.BeginTx(ctx, nil) |
| 122 | if err != nil { |
| 123 | return err |
| 124 | } |
| 125 | defer tx.Rollback() |
| 126 | if err := fn(tx); err != nil { |
| 127 | return err |
| 128 | } |
| 129 | return tx.Commit() |
| 130 | } |