diff --git a/clientdb/db.go b/clientdb/db.go index 63a8f0b..bd77d8c 100644 --- a/clientdb/db.go +++ b/clientdb/db.go @@ -101,7 +101,10 @@ func initDB(filepath string, firstInit bool) (*bbolt.DB, error) { if err != nil { return err } - + _, err = tx.CreateBucketIfNotExists(eventBucketKey) + if err != nil { + return err + } snapshotBucket, err := tx.CreateBucketIfNotExists( batchSnapshotBucketKey, ) diff --git a/clientdb/metadata.go b/clientdb/metadata.go index c49ec5c..82e5663 100644 --- a/clientdb/metadata.go +++ b/clientdb/metadata.go @@ -6,6 +6,7 @@ import ( "fmt" "github.com/btcsuite/btcwallet/wtxmgr" + "github.com/lightninglabs/pool/clientdb/migrations" "go.etcd.io/bbolt" ) @@ -23,7 +24,7 @@ var ( // database version. dbVersionKey = []byte("version") - // lockIDKey is the daabase key used for storing/retrieving the global + // lockIDKey is the database key used for storing/retrieving the global // lock ID to use when leasing outputs from the backing lnd node's // wallet. This is mostly required so that calls to LeaseOutput are // idempotent when attempting to lease an output we already have a lease @@ -38,9 +39,11 @@ var ( // of database don't match with latest version this list will be used // for retrieving all migration function that are need to apply to the // current db. - migrations []migration + dbVersions = []migration{ + migrations.AddInitialOrderTimestamps, + } - latestDBVersion = uint32(len(migrations)) + latestDBVersion = uint32(len(dbVersions)) ) // getDBVersion retrieves the current database version. @@ -132,7 +135,7 @@ func syncVersions(db *bbolt.DB) error { for v := currentVersion; v < latestDBVersion; v++ { log.Infof("Applying migration #%v", v+1) - migration := migrations[v] + migration := dbVersions[v] if err := migration(tx); err != nil { log.Infof("Unable to apply migration #%v", v+1) return err diff --git a/clientdb/migrations/migration_01_add_events.go b/clientdb/migrations/migration_01_add_events.go new file mode 100644 index 0000000..844e236 --- /dev/null +++ b/clientdb/migrations/migration_01_add_events.go @@ -0,0 +1,130 @@ +package migrations + +import ( + "bytes" + "encoding/binary" + "fmt" + "time" + + "github.com/lightninglabs/pool/event" + "go.etcd.io/bbolt" +) + +var ( + // ordersBucketKey is a bucket that contains all orders that are + // currently pending or completed. This bucket is keyed by the nonce and + // leads to a nested sub-bucket that houses information for that order. + ordersBucketKey = []byte("orders") + + // eventBucketKey is the top level bucket where we can find all events + // of the system. These events are indexed by their timestamps which are + // guaranteed to be unique by the clientdb API. + eventBucketKey = []byte("event") + + // eventRefSubBucket is the sub bucket for event references. We only + // store a reference to the event's timestamp and type in the event + // "owner"'s sub bucket (for example the sub bucket of the order the + // event belongs to). + eventRefSubBucket = []byte("event-ref") + + // byteOrder is the default byte order we'll use for serialization + // within the database. + byteOrder = binary.BigEndian +) + +// AddInitialOrderTimestamps creates an "order created" timestamp for each of +// the existing orders. This will only be executed once, when the main event +// bucket is created for the first time. +func AddInitialOrderTimestamps(tx *bbolt.Tx) error { + // We back date all existing orders to the beginning of the year of hell + // to signal this isn't the real timestamp it was created at but + // something we added later. + fixedTimestamp := time.Date( + 2020, time.January, 1, 0, 0, 0, 0, time.UTC, + ).UnixNano() + + // First, we'll grab our main order bucket key. + ordersBucket := tx.Bucket(ordersBucketKey) + if ordersBucket == nil { + return fmt.Errorf("bucket \"%v\" does not exist", + string(ordersBucketKey)) + } + + // We'll now traverse the root bucket for all orders. The primary key is + // the order nonce itself. We create a new event for each order with a + // timestamp that's unique for each event. + var ( + idx int64 + tsScratchSpace [event.TimestampLength]byte + eventType = byte(event.TypeOrderCreated) + eventRefs = make(map[[32]byte]uint64) + ) + err := ordersBucket.ForEach(func(nonceBytes, val []byte) error { + // Only go into things that we know are sub-bucket keys. + if val != nil { + return nil + } + + // Get the order nonce and make sure we can add the initial + // creation event for that order. + var nonce [32]byte + copy(nonce[:], nonceBytes) + orderBucket := ordersBucket.Bucket(nonce[:]) + if orderBucket == nil { + return fmt.Errorf("order bucket not found") + } + + // First of all, encode the timestamp to its binary value and + // keep track of it so we can create a reference in the order's + // event reference sub bucket later. + ts := uint64(fixedTimestamp + idx) + eventRefs[nonce] = ts + byteOrder.PutUint64(tsScratchSpace[:], ts) + + // With the key encoded, we'll then encode the event into our + // buffer, then write it out to disk. + evtBucket := tx.Bucket(eventBucketKey) + var eventBuf bytes.Buffer + if err := eventBuf.WriteByte(eventType); err != nil { + return err + } + if _, err := eventBuf.Write(nonceBytes); err != nil { + return err + } + err := evtBucket.Put(tsScratchSpace[:], eventBuf.Bytes()) + if err != nil { + return err + } + + idx++ + + return nil + }) + if err != nil { + return err + } + + // Because we can't modify the order bucket in the ForEach above + // directly, we need to insert the references outside of the callback. + for nonce, ts := range eventRefs { + orderBucket := ordersBucket.Bucket(nonce[:]) + if orderBucket == nil { + return fmt.Errorf("order bucket not found") + } + + // We also need to store a reference entry in the order bucket. + eventSubBucket, err := orderBucket.CreateBucketIfNotExists( + eventRefSubBucket, + ) + if err != nil { + return err + } + byteOrder.PutUint64(tsScratchSpace[:], ts) + err = eventSubBucket.Put(tsScratchSpace[:], []byte{eventType}) + if err != nil { + return err + } + } + + return nil +}