cmd/sb/sb2/commands/dump.go

package commands

import (
	"fmt"
	"io"
	"strconv"
	"strings"

	"v.io/v23/context"
	"v.io/v23/syncbase"
	"v.io/x/lib/cmdline"
	"v.io/x/ref/cmd/sb/internal/writer"
)

var flagFormat = "table"

func DumpDB(ctx *context.T, env *cmdline.Env, args[] string) error {
	db := ctx.Value("database")
	return dumpDatabase(ctx, env.Stdout, db.(syncbase.Database));
}

func dumpDatabase(ctx *context.T, w io.Writer, d syncbase.Database) error {
	var errors []error
	if err := dumpCollections(ctx, w, d); err != nil {
		errors = append(errors, fmt.Errorf("failed dumping collections: %v", err))
	}
	if err := dumpSyncgroups(ctx, w, d); err != nil {
		errors = append(errors, fmt.Errorf("failed dumping syncgroups: %v", err))
	}
	return mergeErrors(errors)
}

func dumpCollections(ctx *context.T, w io.Writer, d syncbase.Database) error {
	collections, err := d.ListCollections(ctx)
	if err != nil {
		return fmt.Errorf("failed listing collections: %v", err)
	}
	var errs []error
	for _, collection := range collections {
		fmt.Fprintf(w, "collection: %v\n", collection)
		// TODO(ivanpi): Queries currently support only the default user blessing.
		if err := queryExec(ctx, w, d, fmt.Sprintf("select k, v from %s", collection.Name)); err != nil {
			errs = append(errs, fmt.Errorf("> %v: %v", collection, err))
		}
	}
	if len(errs) > 0 {
		return fmt.Errorf("failed dumping %d of %d collections:\n%v", len(errs), len(collections), mergeErrors(errs))
	}
	return nil
}

func dumpSyncgroups(ctx *context.T, w io.Writer, d syncbase.Database) error {
	sgIds, err := d.ListSyncgroups(ctx)
	if err != nil {
		return fmt.Errorf("failed listing syncgroups: %v", err)
	}
	var errs []error
	for _, sgId := range sgIds {
		fmt.Fprintf(w, "syncgroup: %+v\n", sgId)
		if spec, version, err := d.SyncgroupForId(sgId).GetSpec(ctx); err != nil {
			errs = append(errs, err)
		} else {
			fmt.Fprintf(w, "%+v (version: \"%s\")\n", spec, version)
		}
	}
	if len(errs) > 0 {
		return fmt.Errorf("failed dumping %d of %d syncgroups:\n%v", len(errs), len(sgIds), mergeErrors(errs))
	}
	return nil
}

func mergeErrors(errs []error) error {
	if len(errs) == 0 {
		return nil
	}
	err := errs[0]
	for _, e := range errs[1:] {
		err = fmt.Errorf("%v\n%v", err, e)
	}
	return err
}

// Split an error message into an offset and the remaining (i.e., rhs of offset) message.
// The convention for syncql is "<module><optional-rpc>[offset]<remaining-message>".
func splitError(err error) (int64, string) {
	errMsg := err.Error()
	idx1 := strings.Index(errMsg, "[")
	idx2 := strings.Index(errMsg, "]")
	if idx1 == -1 || idx2 == -1 {
		return 0, errMsg
	}
	offsetString := errMsg[idx1+1 : idx2]
	offset, err := strconv.ParseInt(offsetString, 10, 64)
	if err != nil {
		return 0, errMsg
	}
	return offset, errMsg[idx2+1:]
}

func queryExec(ctx *context.T, w io.Writer, d syncbase.Database, q string) error {
	columnNames, rs, err := d.Exec(ctx, q)
	if err != nil {
		off, msg := splitError(err)
		return fmt.Errorf("\n%s\n%s^\n%d: %s", q, strings.Repeat(" ", int(off)), off+1, msg)
	}
	return writer.NewTableWriter(w).Write(columnNames, rs)
	//return flagFormat.NewWriter(w).Write(columnNames, rs)
}