Rename binary to litestream; add replicate command

cmd/litestream/config.go

@@ -0,0 +1,69 @@
+package main
+
+import (
+	"flag"
+	"fmt"
+	"io/ioutil"
+	"os"
+	"os/user"
+	"path/filepath"
+	"strings"
+
+	"gopkg.in/yaml.v2"
+)
+
+// Default settings.
+const (
+	DefaultConfigPath = "~/litestream.yml"
+)
+
+// Config represents a configuration file for the litestream daemon.
+type Config struct {
+	DBs []*DBConfig `yaml:"databases"`
+}
+
+// DefaultConfig returns a new instance of Config with defaults set.
+func DefaultConfig() Config {
+	return Config{}
+}
+
+// ReadConfigFile unmarshals config from filename. Expands path if needed.
+func ReadConfigFile(filename string) (Config, error) {
+	config := DefaultConfig()
+
+	// Expand filename, if necessary.
+	if prefix := "~" + string(os.PathSeparator); strings.HasPrefix(filename, prefix) {
+		u, err := user.Current()
+		if err != nil {
+			return config, err
+		} else if u.HomeDir == "" {
+			return config, fmt.Errorf("home directory unset")
+		}
+		filename = filepath.Join(u.HomeDir, strings.TrimPrefix(filename, prefix))
+	}
+
+	// Read & deserialize configuration.
+	if buf, err := ioutil.ReadFile(filename); os.IsNotExist(err) {
+		return config, fmt.Errorf("config file not found: %s", filename)
+	} else if err != nil {
+		return config, err
+	} else if err := yaml.Unmarshal(buf, &config); err != nil {
+		return config, err
+	}
+	return config, nil
+}
+
+type DBConfig struct {
+	Path     string           `yaml:"path"`
+	Replicas []*ReplicaConfig `yaml:"replicas"`
+}
+
+type ReplicaConfig struct {
+	Type string `yaml:"type"` // "file", "s3"
+	Name string `yaml:"name"` // name of replicator, optional.
+	Path string `yaml:"path"` // used for file replicators
+}
+
+func registerConfigFlag(fs *flag.FlagSet, p *string) {
+	fs.StringVar(p, "config", DefaultConfigPath, "config path")
+}

cmd/litestream/main.go

@@ -0,0 +1,68 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"log"
+	"os"
+	"strings"
+)
+
+// Build information.
+var (
+	Version = "(development build)"
+)
+
+func main() {
+	log.SetFlags(0)
+
+	m := NewMain()
+	if err := m.Run(context.Background(), os.Args[1:]); err == flag.ErrHelp {
+		os.Exit(1)
+	} else if err != nil {
+		fmt.Fprintln(os.Stderr, err)
+		os.Exit(1)
+	}
+}
+
+type Main struct{}
+
+func NewMain() *Main {
+	return &Main{}
+}
+
+func (m *Main) Run(ctx context.Context, args []string) (err error) {
+	var cmd string
+	if len(args) > 0 {
+		cmd, args = args[0], args[1:]
+	}
+
+	switch cmd {
+	case "replicate":
+		return (&ReplicateCommand{}).Run(ctx, args)
+	case "version":
+		return (&VersionCommand{}).Run(ctx, args)
+	default:
+		if cmd == "" || cmd == "help" || strings.HasPrefix(cmd, "-") {
+			m.Usage()
+			return flag.ErrHelp
+		}
+		return fmt.Errorf("litestream %s: unknown command", cmd)
+	}
+}
+
+func (m *Main) Usage() {
+	fmt.Println(`
+litestream is a tool for replicating SQLite databases.
+
+Usage:
+
+	litestream <command> [arguments]
+
+The commands are:
+
+	replicate   runs a server to replicate databases
+	version     prints the version
+`[1:])
+}

cmd/litestream/replicate.go

@@ -0,0 +1,148 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"flag"
+	"fmt"
+	"os"
+	"os/signal"
+
+	"github.com/benbjohnson/litestream"
+)
+
+type ReplicateCommand struct {
+	ConfigPath string
+	Config     Config
+
+	// List of managed databases specified in the config.
+	DBs []*litestream.DB
+}
+
+func NewReplicateCommand() *ReplicateCommand {
+	return &ReplicateCommand{}
+}
+
+// Run loads all databases specified in the configuration.
+func (c *ReplicateCommand) Run(ctx context.Context, args []string) (err error) {
+	fs := flag.NewFlagSet("litestream-replicate", flag.ContinueOnError)
+	registerConfigFlag(fs, &c.ConfigPath)
+	fs.Usage = c.Usage
+	if err := fs.Parse(args); err != nil {
+		return err
+	}
+
+	// Load configuration.
+	if c.ConfigPath == "" {
+		return errors.New("-config required")
+	}
+	config, err := ReadConfigFile(c.ConfigPath)
+	if err != nil {
+		return err
+	}
+
+	// Setup signal handler.
+	ctx, cancel := context.WithCancel(context.Background())
+	ch := make(chan os.Signal, 1)
+	signal.Notify(ch, os.Interrupt)
+	go func() { <-ch; cancel() }()
+
+	// Display version information.
+	fmt.Printf("litestream %s\n", Version)
+
+	if len(config.DBs) == 0 {
+		return errors.New("configuration must specify at least one database")
+	}
+
+	for _, dbc := range config.DBs {
+		if err := c.openDB(dbc); err != nil {
+			return err
+		}
+	}
+
+	// Notify user that initialization is done.
+	fmt.Printf("Initialized with %d databases.\n", len(c.DBs))
+
+	// Wait for signal to stop program.
+	<-ctx.Done()
+	signal.Reset()
+
+	// Gracefully close
+	if err := c.Close(); err != nil {
+		fmt.Fprintln(os.Stderr, err)
+		os.Exit(1)
+	}
+
+	return nil
+}
+
+// openDB instantiates and initializes a DB based on a configuration.
+func (c *ReplicateCommand) openDB(config *DBConfig) error {
+	// Initialize database with given path.
+	db := litestream.NewDB(config.Path)
+
+	// Instantiate and attach replicators.
+	for _, rconfig := range config.Replicas {
+		r, err := c.createReplicator(db, rconfig)
+		if err != nil {
+			return err
+		}
+		db.Replicators = append(db.Replicators, r)
+	}
+
+	// Open database & attach to program.
+	if err := db.Open(); err != nil {
+		return err
+	}
+	c.DBs = append(c.DBs, db)
+
+	return nil
+}
+
+// createReplicator instantiates a replicator for a DB based on a config.
+func (c *ReplicateCommand) createReplicator(db *litestream.DB, config *ReplicaConfig) (litestream.Replicator, error) {
+	switch config.Type {
+	case "", "file":
+		return c.createFileReplicator(db, config)
+	default:
+		return nil, fmt.Errorf("unknown replicator type in config: %q", config.Type)
+	}
+}
+
+// createFileReplicator returns a new instance of FileReplicator build from config.
+func (c *ReplicateCommand) createFileReplicator(db *litestream.DB, config *ReplicaConfig) (*litestream.FileReplicator, error) {
+	if config.Path == "" {
+		return nil, fmt.Errorf("file replicator path require for db %q", db.Path())
+	}
+	return litestream.NewFileReplicator(db, config.Name, config.Path), nil
+}
+
+// Close closes all open databases.
+func (c *ReplicateCommand) Close() (err error) {
+	for _, db := range c.DBs {
+		if e := db.SoftClose(); e != nil {
+			fmt.Printf("error closing db: path=%s err=%s\n", db.Path(), e)
+			if err == nil {
+				err = e
+			}
+		}
+	}
+	return err
+}
+
+func (c *ReplicateCommand) Usage() {
+	fmt.Println(`
+The replicate command starts a server to monitor & replicate databases 
+specified in your configuration file.
+
+Usage:
+
+	litestream replicate [arguments]
+
+Arguments:
+
+	-config PATH
+	    Specifies the configuration file. Defaults to ~/litestream.yml
+
+`[1:])
+}

cmd/litestream/version.go

@@ -0,0 +1,31 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+)
+
+type VersionCommand struct{}
+
+func (c *VersionCommand) Run(ctx context.Context, args []string) (err error) {
+	fs := flag.NewFlagSet("litestream-version", flag.ContinueOnError)
+	fs.Usage = c.Usage
+	if err := fs.Parse(args); err != nil {
+		return err
+	}
+
+	fmt.Println("litestream", Version)
+
+	return nil
+}
+
+func (c *VersionCommand) Usage() {
+	fmt.Println(`
+Prints the version.
+
+Usage:
+
+	litestream version
+`[1:])
+}