Add metamorph records clearing

background_worker/jobs/clear_block_transactions_map.go

@@ -9,9 +9,10 @@ import (
 	_ "github.com/lib/pq"
 )
 
-func (c ClearJob) ClearBlockTransactionsMap(params ClearRecrodsParams) error {
+func (c ClearJob) ClearBlockTransactionsMap(params ClearRecordsParams) error {
 	Log(INFO, "Connecting to database ...")
 	conn, err := sqlx.Open(params.Scheme(), params.String())
+
 	if err != nil {
 		Log(ERROR, fmt.Sprintf("unable to create connection %s", err))
 		return err

background_worker/jobs/clear_block_transactions_map_test.go

@@ -16,11 +16,11 @@ import (
 )
 
 type ClearBlockTransactionsMapSuite struct {
-	DatabaseTestSuite
+	BlockTXDBTestSuite
 }
 
 func (s *ClearBlockTransactionsMapSuite) Test() {
-	params := ClearRecrodsParams{
+	params := ClearRecordsParams{
 		DBConnectionParams:  DefaultParams,
 		RecordRetentionDays: 10,
 	}

background_worker/jobs/clear_blocks.go

@@ -14,7 +14,7 @@ const (
 	numericalDateHourLayout = "2006010215"
 )
 
-type ClearRecrodsParams struct {
+type ClearRecordsParams struct {
 	dbconn.DBConnectionParams
 	RecordRetentionDays int
 }
@@ -41,7 +41,7 @@ func NewClearJob(opts ...func(job *ClearJob)) *ClearJob {
 	return c
 }
 
-func (c ClearJob) ClearBlocks(params ClearRecrodsParams) error {
+func (c ClearJob) ClearBlocks(params ClearRecordsParams) error {
 	Log(INFO, "Connecting to database ...")
 
 	conn, err := sqlx.Open(params.Scheme(), params.String())

background_worker/jobs/clear_blocks_test.go

@@ -16,11 +16,11 @@ import (
 )
 
 type ClearJobSuite struct {
-	DatabaseTestSuite
+	BlockTXDBTestSuite
 }
 
 func (s *ClearJobSuite) Test() {
-	params := ClearRecrodsParams{
+	params := ClearRecordsParams{
 		DBConnectionParams:  DefaultParams,
 		RecordRetentionDays: 10,
 	}

background_worker/jobs/clear_metamorph.go

@@ -0,0 +1,43 @@
+package jobs
+
+import (
+	"fmt"
+
+	"github.com/jmoiron/sqlx"
+)
+
+func runDelete(table string, params ClearRecordsParams) error {
+	Log(INFO, "Connecting to database ...")
+	conn, err := sqlx.Open(params.Scheme(), params.String())
+	if err != nil {
+		Log(ERROR, fmt.Sprintf("unable to create connection %s", err))
+		return err
+	}
+	interval := fmt.Sprintf("%d days", params.RecordRetentionDays)
+
+	stmt, err := conn.Preparex(fmt.Sprintf("DELETE FROM %s WHERE inserted_at <= (CURRENT_DATE - $1::interval)", table))
+	if err != nil {
+		Log(ERROR, fmt.Sprintf("unable to prepare statement %s", err))
+		return err
+	}
+
+	res, err := stmt.Exec(interval)
+	if err != nil {
+		Log(ERROR, "unable to delete block rows")
+		return err
+	}
+	rows, _ := res.RowsAffected()
+	Log(INFO, fmt.Sprintf("Successfully deleted %d rows", rows))
+	return nil
+}
+
+func ClearMetamorph(params ClearRecordsParams) error {
+	if err := runDelete("metamorph.blocks", params); err != nil {
+		return err
+	}
+	if err := runDelete("metamorph.transactions", params); err != nil {
+		return err
+	}
+
+	return nil
+}

background_worker/jobs/clear_metamorph_test.go

@@ -0,0 +1,71 @@
+package jobs
+
+import (
+	"testing"
+	"time"
+
+	. "github.com/bitcoin-sv/arc/database_testing"
+	"github.com/bitcoin-sv/arc/metamorph/store"
+	"github.com/jmoiron/sqlx"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/stretchr/testify/suite"
+)
+
+type ClearMetamorphSuite struct {
+	MetamorphDBTestSuite
+}
+
+func (s *ClearMetamorphSuite) Test() {
+
+	for i := 0; i < 5; i++ {
+		blk := GetTestMMBlock()
+		blk.InsertedAt = time.Now().Add(-20 * 24 * time.Hour)
+		s.InsertBlock(blk)
+	}
+
+	for i := 0; i < 5; i++ {
+		blk := GetTestMMBlock()
+		blk.InsertedAt = time.Now().Add(-1 * 24 * time.Hour)
+		s.InsertBlock(blk)
+	}
+
+	for i := 0; i < 5; i++ {
+		tx := GetTestMMTransaction()
+		tx.InsertedAt = time.Now().Add(-20 * 24 * time.Hour)
+
+		s.InsertTransaction(tx)
+	}
+
+	for i := 0; i < 5; i++ {
+		tx := GetTestMMTransaction()
+		tx.InsertedAt = time.Now().Add(-1 * 24 * time.Hour)
+
+		s.InsertTransaction(tx)
+	}
+
+	err := ClearMetamorph(ClearRecordsParams{
+		DBConnectionParams:  DefaultMMParams,
+		RecordRetentionDays: 14,
+	})
+
+	require.NoError(s.T(), err)
+
+	db, err := sqlx.Open("postgres", DefaultMMParams.String())
+	require.NoError(s.T(), err)
+
+	var blks []store.Block
+	require.NoError(s.T(), db.Select(&blks, "SELECT * from metamorph.blocks"))
+
+	assert.Len(s.T(), blks, 5)
+
+	var stx []store.Transaction
+	require.NoError(s.T(), db.Select(&stx, "SELECT * from metamorph.transactions"))
+	assert.Len(s.T(), stx, 5)
+
+}
+
+func TestRunClearMM(t *testing.T) {
+	s := new(ClearMetamorphSuite)
+	suite.Run(t, s)
+}

background_worker/jobs/clear_transactions.go

@@ -8,7 +8,7 @@ import (
 	_ "github.com/lib/pq"
 )
 
-func (c ClearJob) ClearTransactions(params ClearRecrodsParams) error {
+func (c ClearJob) ClearTransactions(params ClearRecordsParams) error {
 	Log(INFO, "Connecting to database ...")
 
 	conn, err := sqlx.Open(params.Scheme(), params.String())

background_worker/jobs/clear_transactions_test.go

@@ -16,11 +16,11 @@ import (
 )
 
 type ClearTransactionsSuite struct {
-	DatabaseTestSuite
+	BlockTXDBTestSuite
 }
 
 func (s *ClearTransactionsSuite) Test() {
-	params := ClearRecrodsParams{
+	params := ClearRecordsParams{
 		DBConnectionParams:  DefaultParams,
 		RecordRetentionDays: 10,
 	}

background_worker/scheduler.go

@@ -9,10 +9,10 @@ import (
 type ARCScheduler struct {
 	Scheduler       *gocron.Scheduler
 	IntervalInHours int
-	Params          jobs.ClearRecrodsParams
+	Params          jobs.ClearRecordsParams
 }
 
-func (sched *ARCScheduler) RunJob(table string, job func(params jobs.ClearRecrodsParams) error) {
+func (sched *ARCScheduler) RunJob(table string, job func(params jobs.ClearRecordsParams) error) {
 	_, err := sched.Scheduler.Every(sched.IntervalInHours).Hours().Do(func() {
 		jobs.Log(jobs.INFO, fmt.Sprintf("Clearing expired %s...", table))
 		err := job(sched.Params)

blocktx/store/sql/get_block_for_height_test.go

@@ -13,7 +13,7 @@ import (
 )
 
 type GetBlockByHeightTestSuite struct {
-	DatabaseTestSuite
+	BlockTXDBTestSuite
 }
 
 func (s *GetBlockByHeightTestSuite) Test() {

blocktx/store/sql/get_block_gaps_test.go

@@ -14,7 +14,7 @@ import (
 )
 
 type GetBlockGapTestSuite struct {
-	DatabaseTestSuite
+	BlockTXDBTestSuite
 }
 
 func (s *GetBlockGapTestSuite) Test() {

blocktx/store/sql/get_block_test.go

@@ -14,7 +14,7 @@ import (
 )
 
 type GetBlockTestSuite struct {
-	DatabaseTestSuite
+	BlockTXDBTestSuite
 }
 
 func (s *GetBlockTestSuite) Test() {

blocktx/store/sql/get_block_transactions_test.go

@@ -15,7 +15,7 @@ import (
 )
 
 type GetBlockTransactionsSuite struct {
-	DatabaseTestSuite
+	BlockTXDBTestSuite
 }
 
 func (s *GetBlockTransactionsSuite) Test() {

blocktx/store/sql/get_last_processed_block_test.go

@@ -14,7 +14,7 @@ import (
 )
 
 type GetLastProcessedBlockSuite struct {
-	DatabaseTestSuite
+	BlockTXDBTestSuite
 }
 
 func (s *GetLastProcessedBlockSuite) Test() {

blocktx/store/sql/get_mined_transaction_for_block_test.go

@@ -13,7 +13,7 @@ import (
 )
 
 type MinedTransactionForBlockSuite struct {
-	DatabaseTestSuite
+	BlockTXDBTestSuite
 }
 
 func (s *MinedTransactionForBlockSuite) Test() {

blocktx/store/sql/get_transaction_block_test.go

@@ -13,7 +13,7 @@ import (
 )
 
 type GetTransactionBlockSuite struct {
-	DatabaseTestSuite
+	BlockTXDBTestSuite
 }
 
 func (s *GetTransactionBlockSuite) Test() {

blocktx/store/sql/get_transaction_blocks_test.go

@@ -15,7 +15,7 @@ import (
 )
 
 type GetTransactionBlocksSuite struct {
-	DatabaseTestSuite
+	BlockTXDBTestSuite
 }
 
 func (s *GetTransactionBlocksSuite) Test() {

blocktx/store/sql/get_transaction_merkle_path_test.go

@@ -13,7 +13,7 @@ import (
 )
 
 type GetTransactionMerklePathSuite struct {
-	DatabaseTestSuite
+	BlockTXDBTestSuite
 }
 
 func (s *GetTransactionMerklePathSuite) Test() {

blocktx/store/sql/insert_block_test.go

@@ -14,7 +14,7 @@ import (
 )
 
 type InsertBlockSuite struct {
-	DatabaseTestSuite
+	BlockTXDBTestSuite
 }
 
 func (s *InsertBlockSuite) Test() {

blocktx/store/sql/insert_block_transactions_test.go

@@ -14,7 +14,7 @@ import (
 )
 
 type InsertBlockTransactionsSuite struct {
-	DatabaseTestSuite
+	BlockTXDBTestSuite
 }
 
 type Tx struct {

blocktx/store/sql/orphan_height_test.go

@@ -13,7 +13,7 @@ import (
 )
 
 type OrphanHeightSutie struct {
-	DatabaseTestSuite
+	BlockTXDBTestSuite
 }
 
 func (s *OrphanHeightSutie) Test() {

cmd/background_worker.go

@@ -57,7 +57,7 @@ func StartBackGroundWorker(logger *slog.Logger) (func(), error) {
 		return nil, err
 	}
 
-	params := jobs.ClearRecrodsParams{
+	params := jobs.ClearRecordsParams{
 		DBConnectionParams: dbconn.New(
 			dbHost,
 			dbPort,

cmd/background_worker/main.go

@@ -27,7 +27,7 @@ func main() {
 		}
 	}
 
-	params := jobs.ClearRecrodsParams{
+	params := jobs.ClearRecordsParams{
 		DBConnectionParams: dbconn.New(
 			viper.GetString("cleanBlocks.host"),
 			viper.GetInt("cleanBlocks.port"),

database/migrations/metamorph/postgres/000001_create_transactions.down.sql

@@ -0,0 +1 @@
+DROP TABLE metamorph.transactions;

database/migrations/metamorph/postgres/000001_create_transactions.up.sql

@@ -0,0 +1,20 @@
+CREATE SCHEMA metamorph;
+CREATE TABLE metamorph.transactions (
+    hash BYTEA PRIMARY KEY,
+    stored_at TIMESTAMPTZ,
+    announced_at TIMESTAMPTZ,
+    mined_at TIMESTAMPTZ,
+    status INTEGER,
+    block_height BIGINT,
+    block_hash BYTEA,
+    callback_url TEXT,
+    callback_token TEXT,
+    merkle_proof TEXT,
+    reject_reason TEXT,
+    raw_tx BYTEA,
+    locked_by TEXT,
+    inserted_at_num INTEGER DEFAULT TO_NUMBER(TO_CHAR((NOW()) AT TIME ZONE 'UTC', 'yyyymmddhh24'), '9999999999') NOT NULL
+);
+
+CREATE INDEX ix_metamorph_transactions_locked_by ON metamorph.transactions (locked_by);
+CREATE INDEX ix_metamorph_transactions_inserted_at_num ON metamorph.transactions (inserted_at_num);

database/migrations/metamorph/postgres/000002_create_blocks.down.sql

@@ -0,0 +1 @@
+DROP TABLE metamorph.blocks;

database/migrations/metamorph/postgres/000002_create_blocks.up.sql

@@ -0,0 +1,7 @@
+CREATE TABLE metamorph.blocks (
+    hash BYTEA PRIMARY KEY,
+    processed_at TIMESTAMPTZ,
+    inserted_at_num INTEGER DEFAULT TO_NUMBER(TO_CHAR((NOW()) AT TIME ZONE 'UTC', 'yyyymmddhh24'), '9999999999') NOT NULL
+);
+
+CREATE INDEX ix_metamorph_blocks_inserted_at_num ON metamorph.blocks (inserted_at_num);

database/migrations/metamorph/postgres/000003_create_functions.down.sql

@@ -0,0 +1,2 @@
+DROP FUNCTION reverse_bytes_iter(bytes bytea, length int, midpoint int, index int);
+DROP FUNCTION reverse_bytes(bytes bytea);

database/migrations/metamorph/postgres/000003_create_functions.up.sql

@@ -0,0 +1,16 @@
+CREATE OR REPLACE FUNCTION reverse_bytes_iter(bytes bytea, length int, midpoint int, index int)
+	RETURNS bytea AS
+	$$
+SELECT CASE WHEN index >= midpoint THEN bytes ELSE
+    reverse_bytes_iter(
+            set_byte(
+                    set_byte(bytes, index, get_byte(bytes, length-index)),
+                    length-index, get_byte(bytes, index)
+                ),
+            length, midpoint, index + 1
+        )
+           END;
+$$ LANGUAGE SQL IMMUTABLE;
+
+CREATE
+OR REPLACE FUNCTION reverse_bytes(bytes bytea) RETURNS bytea AS 'SELECT reverse_bytes_iter(bytes, octet_length(bytes)-1, octet_length(bytes)/2, 0)' LANGUAGE SQL IMMUTABLE;

database/migrations/metamorph/postgres/000004_create_timestamps.down.sql

@@ -0,0 +1,2 @@
+ALTER TABLE metamorph.blocks DROP COLUMN inserted_at;
+ALTER TABLE metamorph.transactions DROP COLUMN inserted_at;

database/migrations/metamorph/postgres/000004_create_timestamps.up.sql

@@ -0,0 +1,2 @@
+ALTER TABLE metamorph.transactions ADD COLUMN inserted_at TIMESTAMPTZ;
+ALTER TABLE metamorph.blocks ADD COLUMN inserted_at TIMESTAMPTZ;