pocketbase/core/backup_create.go

package core

import (
	"archive/zip"
	"compress/flate"
	"context"
	"errors"
	"fmt"
	"io"
	"io/fs"
	"log/slog"
	"os"
	"path"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"github.com/pocketbase/dbx"
	"github.com/pocketbase/pocketbase/tools/filesystem"
	"github.com/pocketbase/pocketbase/tools/hook"
	"github.com/pocketbase/pocketbase/tools/security"
	"github.com/pocketbase/pocketbase/tools/store"
)

var errIsDir = errors.New("the specified path is a directory and not a regular file")

// CreateBackup creates a new backup of the current app pb_data directory.
//
// If name is empty, it will be autogenerated.
// If backup with the same name exists, the new backup file will replace it.
//
// To safely perform the backup, it is recommended to have free disk space
// for at least 2x the size of the pb_data directory.
//
// By default backups are stored in pb_data/backups
// (the backups directory itself is excluded from the generated backup).
//
// Backups can be stored on S3 if it is configured in app.Settings().Backups.
// When using S3 storage for the uploaded collection files, you have to
// take care manually to backup those since they are not part of the pb_data.
//
// DB write locks are minimal and isolated only for the duration of the
// VACUUM INTO statement that creates a live copy of the app database.
//
// The backup works as follow:
//
//  1. Start listening for DELETED storage files.
//     In case a file is being deleted while the backup is still ongoing,
//     we directly copy it in the zip before the deletion and mark it as "excluded".
//
//  2. Copy the main database with VACUUM INTO, write it in the zip and mark it as "excluded".
//
//  3. Stop listening for DELETED files.
//
//  4. Start listening for NEW storage files and mark all new files from this point as "excluded".
//
//  5. Copy the logs database with VACUUM INTO, write it in the zip and mark it as "excluded".
//
//  6. Copy the rest of the pb_data files in the zip while ignoring the "excluded" list (it should be concurrent safe).
//
//  7. Stop listening for NEW storage files.
//
// While there is a risk for a race condition between steps 1, 2 and 3, it is an
// acceptable trade-off between performance and correctness because in
// the worst case there will be some unused storage files in the backup that don't do any harm.
func (app *BaseApp) CreateBackup(ctx context.Context, name string) error {
	if app.Store().Has(StoreKeyActiveBackup) {
		return errors.New("try again later - another backup/restore operation has already been started")
	}

	app.Store().Set(StoreKeyActiveBackup, name)
	defer app.Store().Remove(StoreKeyActiveBackup)

	startTime := time.Now()

	event := new(BackupEvent)
	event.App = app
	event.Context = ctx
	event.Name = name
	// default root dir entries to exclude from the backup generation
	event.Exclude = []string{
		LocalBackupsDirName,
		LocalTempDirName,
		LocalNotifyDirName,
		LocalAutocertCacheDirName,
		lostFoundDirName,
	}

	return app.OnBackupCreate().Trigger(event, func(e *BackupEvent) error {
		if e.Name == "" {
			e.Name = generateBackupName(e.App, "pb_backup_")
		}

		app.Logger().Debug("[" + e.Name + "] zip archive started")

		// create backup zip
		// (it needs to be inside the current pb_data to avoid "cross-device link" errors)
		// -----------------------------------------------------------
		tempZipPath := filepath.Join(app.DataDir(), LocalTempDirName, "pb_backup_"+security.PseudorandomString(6))
		err := createZip(e, tempZipPath)
		if err != nil {
			return err
		}
		defer os.Remove(tempZipPath)

		// persist the backup in the backups filesystem
		// -----------------------------------------------------------
		fsys, err := e.App.NewBackupsFilesystem()
		if err != nil {
			return err
		}
		defer fsys.Close()

		fsys.SetContext(e.Context)

		file, err := filesystem.NewFileFromPath(tempZipPath)
		if err != nil {
			return err
		}
		file.OriginalName = e.Name
		file.Name = file.OriginalName

		err = fsys.UploadFile(file, file.Name)
		if err != nil {
			return err
		}

		app.Logger().Debug(
			"["+e.Name+"] zip archive completed",
			slog.Float64("execTime", float64(time.Since(startTime))/float64(time.Millisecond)),
		)

		return nil
	})
}

func createZip(be *BackupEvent, tempZipPath string) error {
	logPrefix := "[" + be.Name + "] "

	// make sure that the special temp directory exists
	localTempDir := filepath.Dir(tempZipPath)
	if err := os.MkdirAll(localTempDir, os.ModePerm); err != nil {
		return fmt.Errorf(logPrefix+"failed to create temp dir: %w", err)
	}

	const tempFilesHookId = "__pbTempBackupFilesystemWatcher__"
	defer func() {
		// unbind again in cacase of an error
		be.App.onFilesystemDelete().Unbind(tempFilesHookId)
		be.App.onFilesystemNewWriter().Unbind(tempFilesHookId)
	}()

	zipper, err := newZipWriter(tempZipPath)
	if err != nil {
		return err
	}
	defer zipper.close()

	excluded := store.New[string, struct{}](nil)
	for _, name := range be.Exclude {
		excluded.Set(normalizePathExclude(name), struct{}{})
	}

	// init deleted files tracker
	// ---------------------------------------------------------------
	be.App.onFilesystemDelete().Bind(&hook.Handler[*FilesystemDeleteEvent]{
		Id:       tempFilesHookId,
		Priority: -99,
		Func: func(e *FilesystemDeleteEvent) error {
			// note: the zip header name allow only forward slashes
			zipPath := path.Join(LocalStorageDirName, e.FileKey)

			if excluded.Has(normalizePathExclude(zipPath)) || be.App.Settings().S3.Enabled {
				return e.Next()
			}

			localPath := filepath.Join(
				be.App.DataDir(),
				LocalStorageDirName,
				e.FileKey,
			)

			// copy to zip before delete
			err := zipper.copyFileToZip(localPath, zipPath)
			if err != nil {
				// it is ok to ignore directories
				if !errors.Is(err, errIsDir) {
					be.App.Logger().Warn(
						logPrefix+"failed to copy file in backup zip before delete",
						slog.Any("error", err),
						slog.String("file", e.FileKey),
					)
				}
			} else {
				// mark that it was already copied
				excluded.Set(normalizePathExclude(zipPath), struct{}{})
			}

			// proceed with the normal deletion
			return e.Next()
		},
	})

	// copy data.db
	// ---------------------------------------------------------------
	dataStartTime := time.Now()
	tempDataDBPath := filepath.Join(localTempDir, dataDBFilename)

	_, err = be.App.ConcurrentDB().NewQuery("VACUUM INTO {:path}").Bind(dbx.Params{"path": tempDataDBPath}).Execute()
	if err != nil {
		return err
	}

	// eagerly stop listening for deleted files since we already have what we needed
	be.App.onFilesystemDelete().Unbind(tempFilesHookId)

	be.App.Logger().Debug(
		logPrefix+dataDBFilename+" copy completed",
		slog.Float64("execTime", float64(time.Since(dataStartTime))/float64(time.Millisecond)),
	)

	err = zipper.copyFileToZip(tempDataDBPath, dataDBFilename)
	if err != nil {
		_ = os.Remove(tempDataDBPath)
		return err
	}
	_ = os.Remove(tempDataDBPath)

	excluded.Set(normalizePathExclude(dataDBFilename), struct{}{})
	excluded.Set(normalizePathExclude(dataDBFilename+"-wal"), struct{}{})
	excluded.Set(normalizePathExclude(dataDBFilename+"-shm"), struct{}{})

	// init to-be-created files tracker
	// ---------------------------------------------------------------
	be.App.onFilesystemNewWriter().Bind(&hook.Handler[*FilesystemNewWriterEvent]{
		Id:       tempFilesHookId,
		Priority: -99,
		Func: func(e *FilesystemNewWriterEvent) error {
			if !be.App.Settings().S3.Enabled {
				// mark for exclude even if the writer eventually fails
				// (all record files have random name so collisions are unlikely)
				name := normalizePathExclude(filepath.Join(LocalStorageDirName, e.FileKey))
				excluded.Set(name, struct{}{})
			}

			return e.Next()
		},
	})

	// copy auxiliary.db
	// ---------------------------------------------------------------
	auxStartTime := time.Now()
	tempAuxDBPath := filepath.Join(localTempDir, auxDBFilename)

	_, err = be.App.AuxConcurrentDB().NewQuery("VACUUM INTO {:path}").Bind(dbx.Params{"path": tempAuxDBPath}).Execute()
	if err != nil {
		return err
	}

	be.App.Logger().Debug(
		logPrefix+auxDBFilename+" copy completed",
		slog.Float64("execTime", float64(time.Since(auxStartTime))/float64(time.Millisecond)),
	)

	err = zipper.copyFileToZip(tempAuxDBPath, auxDBFilename)
	if err != nil {
		_ = os.Remove(tempAuxDBPath)
		return err
	}
	_ = os.Remove(tempAuxDBPath)

	excluded.Set(normalizePathExclude(auxDBFilename), struct{}{})
	excluded.Set(normalizePathExclude(auxDBFilename+"-wal"), struct{}{})
	excluded.Set(normalizePathExclude(auxDBFilename+"-shm"), struct{}{})

	// try to run manual checkpoints to ensure that all wal writes during the
	// previous VACUUM INTO are transferred and don't accumulate
	// (errors are ignore because some drivers may not support the wal_checkpoint pragma)
	// ---------------------------------------------------------------
	_, _ = be.App.NonconcurrentDB().NewQuery("PRAGMA wal_checkpoint(TRUNCATE)").Execute()
	_, _ = be.App.AuxNonconcurrentDB().NewQuery("PRAGMA wal_checkpoint(TRUNCATE)").Execute()

	// copy the rest of the pb_data
	// ---------------------------------------------------------------
	err = zipper.copyDirToZip(os.DirFS(be.App.DataDir()), excluded)
	if err != nil {
		return err
	}

	return zipper.close()
}

// normalize the provided file path to always end with forward slash
func normalizePathExclude(filePath string) string {
	return path.Clean(filePath) + "/"
}

type zipWriter struct {
	mu     sync.Mutex
	w      *zip.Writer
	f      *os.File
	closed bool
}

func newZipWriter(zipFilePath string) (*zipWriter, error) {
	f, err := os.Create(zipFilePath)
	if err != nil {
		return nil, err
	}

	w := zip.NewWriter(f)
	w.RegisterCompressor(zip.Deflate, func(out io.Writer) (io.WriteCloser, error) {
		return flate.NewWriter(out, flate.BestSpeed)
	})

	return &zipWriter{
		w: w,
		f: f,
	}, nil
}

func (z *zipWriter) close() error {
	z.mu.Lock()
	defer z.mu.Unlock()

	if z.closed {
		return nil
	}

	z.closed = true

	return errors.Join(z.w.Close(), z.f.Close())
}

func (z *zipWriter) copyFileToZip(localPath string, zipPath string) error {
	info, err := os.Stat(localPath)
	if err != nil {
		return err
	}

	if info.IsDir() {
		return errIsDir
	}

	h, err := zip.FileInfoHeader(info)
	if err != nil {
		return err
	}

	h.Name = zipPath
	h.Method = zip.Deflate

	z.mu.Lock()
	defer z.mu.Unlock()

	if z.closed {
		return errors.New("zip writer is already closed")
	}

	fw, err := z.w.CreateHeader(h)
	if err != nil {
		return err
	}

	f, err := os.Open(localPath)
	if err != nil {
		return err
	}
	defer f.Close()

	_, err = io.Copy(fw, f)

	return err
}

func (z *zipWriter) copyDirToZip(fsys fs.FS, excludedPrefixes *store.Store[string, struct{}]) error {
	return fs.WalkDir(fsys, ".", func(name string, d fs.DirEntry, err error) error {
		if err != nil {
			return err
		}

		// skip excluded prefixes
		if excludedPrefixes != nil {
			check := normalizePathExclude(name)
			prefixes := excludedPrefixes.Keys() // refetch to avoid races
			for _, prefix := range prefixes {
				if strings.HasPrefix(check, prefix) {
					if d.IsDir() {
						return fs.SkipDir
					}
					return nil
				}
			}
		}

		if d.IsDir() {
			return nil
		}

		info, err := d.Info()
		if err != nil {
			return err
		}

		h, err := zip.FileInfoHeader(info)
		if err != nil {
			return err
		}

		h.Name = name
		h.Method = zip.Deflate

		z.mu.Lock()
		defer z.mu.Unlock()

		if z.closed {
			// note: fs.WalkDir perform direct comparison with the value
			return fs.SkipAll
		}

		fw, err := z.w.CreateHeader(h)
		if err != nil {
			return err
		}

		f, err := fsys.Open(name)
		if err != nil {
			return err
		}
		defer f.Close()

		_, err = io.Copy(fw, f)

		return err
	})
}