1package replication23import (4 "bytes"5 "context"6 "crypto/sha256"7 "database/sql"8 "encoding/base64"9 "encoding/hex"10 "encoding/json"11 "errors"12 "fmt"13 "io"14 "strings"15 "time"1617 "github.com/FelineStateMachine/tinyrelay/internal/event"18 "github.com/FelineStateMachine/tinyrelay/internal/storage"19)2021// BackupFormat identifies the JSON archive format produced by this package.22const BackupFormat = "tinyrelay/backup/1"2324// BackupArchive is a sealed event and application-state export.25type BackupArchive struct {26 Format string `json:"format"`27 Events []event.Event `json:"events"`28 State BackupState `json:"state"`29 SHA256 string `json:"sha256"`30 Created int64 `json:"created_at"`31}3233// BackupState contains application-owned state and exported objects.34type BackupState struct {35 Config json.RawMessage `json:"config,omitempty"`36 Community json.RawMessage `json:"community,omitempty"`37 Identity json.RawMessage `json:"identity,omitempty"`38 Blobs []BackupObject `json:"blobs,omitempty"`39 Sites []BackupObject `json:"sites,omitempty"`40 Git []BackupObject `json:"git,omitempty"`41 GitSQL json.RawMessage `json:"sqlGit,omitempty"`42 Database []BackupTable `json:"database,omitempty"`43}4445// BackupTable is a named table snapshot represented as JSON rows.46type BackupTable struct {47 Name string `json:"name"`48 Columns []string `json:"columns"`49 Rows [][]json.RawMessage `json:"rows"`50}5152// BackupObject is an exported named file or blob with an integrity digest.53type BackupObject struct {54 Name string `json:"name"`55 SHA256 string `json:"sha256"`56 Content []byte `json:"content"`57 Type string `json:"type,omitempty"`58 Uploader string `json:"uploader,omitempty"`59 Uploaded int64 `json:"uploaded,omitempty"`60}6162// BackupProvider snapshots and restores application state associated with a63// relay database. The provider owns its external resources.64type BackupProvider interface {65 Snapshot(ctx context.Context) (BackupState, error)66 Restore(ctx context.Context, state BackupState) error67}6869// FullBackupProvider can produce a complete archive using its own snapshot70// coordination.71type FullBackupProvider interface {72 BackupProvider73 SnapshotArchive(ctx context.Context, created int64) (BackupArchive, error)74}7576// RestoreFinalizer runs application cleanup after a successful restore.77type RestoreFinalizer interface {78 FinalizeRestore(ctx context.Context) error79}8081func CreateBackup(ctx context.Context, store *storage.Store, now int64) (BackupArchive, error) {82 rows, err := store.After(ctx, 0, event.Filter{Tags: map[string][]string{}}, storage.QueryOptions{Now: now, Access: storage.Access{All: true}, Limit: 0})83 if err != nil {84 return BackupArchive{}, err85 }86 archive := BackupArchive{Format: BackupFormat, Created: now, Events: make([]event.Event, 0, len(rows))}87 for _, row := range rows {88 archive.Events = append(archive.Events, row.Event)89 }90 return sealBackup(archive)91}9293func CreateBackupWithProvider(ctx context.Context, store *storage.Store, provider BackupProvider, now int64) (BackupArchive, error) {94 if full, ok := provider.(FullBackupProvider); ok {95 return full.SnapshotArchive(ctx, now)96 }97 archive, err := CreateBackup(ctx, store, now)98 if err != nil || provider == nil {99 return archive, err100 }101 state, err := provider.Snapshot(ctx)102 if err != nil {103 return BackupArchive{}, err104 }105 archive.State = state106 return sealBackup(archive)107}108109func SealBackup(archive BackupArchive) (BackupArchive, error) { return sealBackup(archive) }110111func WriteBackup(ctx context.Context, store *storage.Store, output io.Writer, now int64) error {112 archive, err := CreateBackup(ctx, store, now)113 if err != nil {114 return err115 }116 return json.NewEncoder(output).Encode(archive)117}118119func ReadBackup(input io.Reader) (BackupArchive, error) {120 var archive BackupArchive121 decoder := json.NewDecoder(input)122 if err := decoder.Decode(&archive); err != nil {123 return archive, fmt.Errorf("replication: decode backup: %w", err)124 }125 var trailing any126 if err := decoder.Decode(&trailing); err != io.EOF {127 if err == nil {128 return archive, errors.New("replication: trailing JSON after backup")129 }130 return archive, fmt.Errorf("replication: trailing backup data: %w", err)131 }132 if archive.Format != BackupFormat {133 return archive, errors.New("replication: unsupported backup format")134 }135 sealed, err := sealBackup(BackupArchive{Format: archive.Format, Events: archive.Events, State: archive.State, Created: archive.Created})136 if err != nil || sealed.SHA256 != archive.SHA256 {137 return archive, errors.New("replication: backup integrity check failed")138 }139 return archive, nil140}141142// ReadCompatibleBackup accepts both the native archive and bind.ws/relay-backup/1.143// The legacy shape is normalized before restore so callers have one validation path.144func ReadCompatibleBackup(input io.Reader) (BackupArchive, error) {145 data, err := io.ReadAll(input)146 if err != nil {147 return BackupArchive{}, err148 }149 var envelope struct {150 Format string `json:"format"`151 }152 if err := json.Unmarshal(data, &envelope); err != nil {153 return BackupArchive{}, fmt.Errorf("replication: decode backup envelope: %w", err)154 }155 if envelope.Format == BackupFormat {156 return ReadBackup(bytes.NewReader(data))157 }158 if envelope.Format != "bind.ws/relay-backup/1" {159 return BackupArchive{}, errors.New("replication: unsupported backup format")160 }161 return readBindWSBackup(data)162}163164func readBindWSBackup(data []byte) (BackupArchive, error) {165 var legacy struct {166 Format string `json:"format"`167 Manifest struct {168 CreatedAt int64 `json:"createdAt"`169 ArchiveSHA256 string `json:"archiveSha256"`170 Owner string `json:"owner"`171 RelayIdentity string `json:"relayIdentity"`172 Slug string `json:"slug"`173 } `json:"manifest"`174 Config json.RawMessage `json:"config"`175 Events []string `json:"events"`176 Blobs []struct {177 SHA256 string `json:"sha256"`178 Type string `json:"type"`179 Uploader string `json:"uploader"`180 Uploaded int64 `json:"uploaded"`181 Data string `json:"data"`182 } `json:"blobs"`183 Git []struct {184 Key string `json:"key"`185 SHA256 string `json:"sha256"`186 Data string `json:"data"`187 } `json:"git"`188 SQLGit json.RawMessage `json:"sqlGit"`189 State json.RawMessage `json:"state"`190 }191 if err := json.Unmarshal(data, &legacy); err != nil {192 return BackupArchive{}, err193 }194 if legacy.Manifest.ArchiveSHA256 == "" {195 return BackupArchive{}, errors.New("replication: legacy archive checksum missing")196 }197 canonical, err := blankLegacyArchiveHash(data)198 if err != nil || !strings.EqualFold(legacy.Manifest.ArchiveSHA256, digest(canonical)) {199 return BackupArchive{}, errors.New("replication: legacy archive checksum mismatch")200 }201 archive := BackupArchive{Format: BackupFormat, Created: legacy.Manifest.CreatedAt, State: BackupState{Config: legacy.Config, GitSQL: append(json.RawMessage(nil), legacy.SQLGit...)}}202 identity, _ := json.Marshal(map[string]string{"owner": legacy.Manifest.Owner, "relayIdentity": legacy.Manifest.RelayIdentity, "slug": legacy.Manifest.Slug})203 archive.State.Identity = identity204 if len(legacy.State) > 0 && string(legacy.State) != "null" {205 archive.State.Community = append(json.RawMessage(nil), legacy.State...)206 }207 for _, raw := range legacy.Events {208 var item event.Event209 if err := json.Unmarshal([]byte(raw), &item); err != nil {210 return BackupArchive{}, fmt.Errorf("replication: legacy event: %w", err)211 }212 if err := event.Validate(item); err != nil {213 return BackupArchive{}, fmt.Errorf("replication: legacy event validation: %w", err)214 }215 archive.Events = append(archive.Events, item)216 }217 for _, item := range legacy.Blobs {218 content, err := base64.StdEncoding.DecodeString(item.Data)219 if err != nil {220 return BackupArchive{}, err221 }222 if !validDigest(item.SHA256, content) {223 return BackupArchive{}, errors.New("replication: legacy blob checksum mismatch")224 }225 archive.State.Blobs = append(archive.State.Blobs, BackupObject{Name: item.SHA256, SHA256: item.SHA256, Content: content, Type: item.Type, Uploader: item.Uploader, Uploaded: item.Uploaded})226 }227 for _, item := range legacy.Git {228 content, err := base64.StdEncoding.DecodeString(item.Data)229 if err != nil {230 return BackupArchive{}, err231 }232 if item.SHA256 != "" && !validDigest(item.SHA256, content) {233 return BackupArchive{}, errors.New("replication: legacy git checksum mismatch")234 }235 archive.State.Git = append(archive.State.Git, BackupObject{Name: item.Key, SHA256: digest(content), Content: content})236 }237 return sealBackup(archive)238}239240func blankLegacyArchiveHash(data []byte) ([]byte, error) {241 key := []byte(`"archiveSha256"`)242 start := bytes.Index(data, key)243 if start < 0 {244 return nil, errors.New("replication: legacy archive hash field missing")245 }246 colon := bytes.IndexByte(data[start+len(key):], ':')247 if colon < 0 {248 return nil, errors.New("replication: malformed legacy archive hash")249 }250 colon += start + len(key)251 quote := bytes.IndexByte(data[colon+1:], '"')252 if quote < 0 {253 return nil, errors.New("replication: malformed legacy archive hash")254 }255 quote += colon + 1256 end := bytes.IndexByte(data[quote+1:], '"')257 if end < 0 {258 return nil, errors.New("replication: malformed legacy archive hash")259 }260 end += quote + 1261 out := make([]byte, 0, len(data))262 out = append(out, data[:quote+1]...)263 out = append(out, data[end:]...)264 return out, nil265}266267func validDigest(expected string, content []byte) bool {268 return expected != "" && strings.EqualFold(expected, digest(content))269}270func digest(content []byte) string { sum := sha256.Sum256(content); return hex.EncodeToString(sum[:]) }271272func RestoreBackup(ctx context.Context, store *storage.Store, archive BackupArchive, now int64) (int, error) {273 return RestoreBackupWithProvider(ctx, store, nil, archive, now)274}275276func RestoreBackupWithProvider(ctx context.Context, store *storage.Store, provider BackupProvider, archive BackupArchive, now int64) (int, error) {277 if archive.Format != BackupFormat {278 return 0, errors.New("replication: unsupported backup format")279 }280 sealed, err := sealBackup(BackupArchive{Format: archive.Format, Events: archive.Events, State: archive.State, Created: archive.Created})281 if err != nil {282 return 0, err283 }284 if sealed.SHA256 != archive.SHA256 {285 return 0, errors.New("replication: backup integrity check failed")286 }287 for _, item := range archive.Events {288 if err := event.Validate(item); err != nil {289 return 0, fmt.Errorf("replication: restore event validation: %w", err)290 }291 }292 if provider != nil && len(archive.State.Database) > 0 {293 previous, snapshotErr := provider.Snapshot(ctx)294 if snapshotErr != nil {295 return 0, fmt.Errorf("replication: snapshot retained state: %w", snapshotErr)296 }297 if err := provider.Restore(ctx, archive.State); err != nil {298 return 0, fmt.Errorf("replication: restore retained state: %w", err)299 }300 if finalizer, ok := provider.(RestoreFinalizer); ok {301 if err := finalizer.FinalizeRestore(ctx); err != nil {302 rollbackProvider(ctx, provider, previous)303 return 0, fmt.Errorf("replication: finalize restore: %w", err)304 }305 }306 return len(archive.Events), nil307 }308 var previous BackupState309 var havePrevious bool310 if provider != nil {311 previous, err = provider.Snapshot(ctx)312 if err != nil {313 return 0, fmt.Errorf("replication: snapshot retained state: %w", err)314 }315 havePrevious = true316 if err := provider.Restore(ctx, archive.State); err != nil {317 return 0, fmt.Errorf("replication: restore retained state: %w", err)318 }319 }320 stored := 0321 inserted := make([]string, 0, len(archive.Events))322 for _, item := range archive.Events {323 if err := ctx.Err(); err != nil {324 rollbackEvents(ctx, store, inserted)325 if havePrevious {326 rollbackProvider(ctx, provider, previous)327 }328 return stored, err329 }330 if _, err := store.Save(ctx, item, storage.SaveOptions{Now: now}); err != nil {331 if err == storage.ErrDuplicate || err == storage.ErrReplaced {332 continue333 }334 rollbackEvents(ctx, store, inserted)335 if havePrevious {336 rollbackProvider(ctx, provider, previous)337 }338 return stored, fmt.Errorf("replication: restore event: %w", err)339 }340 inserted = append(inserted, item.ID)341 stored++342 }343 if finalizer, ok := provider.(RestoreFinalizer); ok {344 if err := finalizer.FinalizeRestore(ctx); err != nil {345 rollbackEvents(ctx, store, inserted)346 if havePrevious {347 rollbackProvider(ctx, provider, previous)348 }349 return stored, fmt.Errorf("replication: finalize restore: %w", err)350 }351 }352 return stored, nil353}354355func restoreRollbackContext(ctx context.Context) (context.Context, context.CancelFunc) {356 base := context.WithoutCancel(ctx)357 return context.WithTimeout(base, 30*time.Second)358}359360func rollbackEvents(ctx context.Context, store *storage.Store, ids []string) {361 rollback, cancel := restoreRollbackContext(ctx)362 defer cancel()363 rollbackRestoreEvents(rollback, store, ids)364}365366func rollbackProvider(ctx context.Context, provider BackupProvider, state BackupState) {367 if provider == nil {368 return369 }370 rollback, cancel := restoreRollbackContext(ctx)371 defer cancel()372 _ = provider.Restore(rollback, state)373}374375func rollbackRestoreEvents(ctx context.Context, store *storage.Store, ids []string) {376 if len(ids) == 0 || store == nil {377 return378 }379 _ = store.WithTx(ctx, func(tx *sql.Tx) error {380 marks := make([]string, len(ids))381 args := make([]any, len(ids))382 for i, id := range ids {383 marks[i], args[i] = "?", id384 }385 _, err := tx.ExecContext(ctx, `DELETE FROM events WHERE id IN (`+strings.Join(marks, ",")+`)`, args...)386 return err387 })388}389390func sealBackup(archive BackupArchive) (BackupArchive, error) {391 archive.SHA256 = ""392 raw, err := json.Marshal(archive)393 if err != nil {394 return archive, err395 }396 sum := sha256.Sum256(raw)397 archive.SHA256 = hex.EncodeToString(sum[:])398 return archive, nil399}