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,42 @@
+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("blocks", params); err != nil {
+		return err
+	}
+	if err := runDelete("transactions", params); err != nil {
+		return err
+	}
+
+	return nil
+}

background_worker/jobs/clear_metamorph_test.go

@@ -0,0 +1,70 @@
+package jobs
+
+import (
+	. "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"
+	"testing"
+	"time"
+)
+
+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 blocks"))
+
+	assert.Len(s.T(), blks, 5)
+
+	var stx []store.Transaction
+	require.NoError(s.T(), db.Select(&stx, "SELECT * from 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/get_transaction_source_test.go

@@ -13,7 +13,7 @@ import (
 )
 
 type GetTransactionSourceSuite struct {
-	DatabaseTestSuite
+	BlockTXDBTestSuite
 }
 
 func (s *GetTransactionSourceSuite) 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 transactions;

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

 CREATE SCHEMA metamorph; 
 CREATE SCHEMA metamorph; 
@@ -0,0 +1,20 @@
+CREATE TABLE 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    TIMESTAMPTZ
+);
+
+CREATE INDEX ix_metamorph_transactions_locked_by ON transactions (locked_by);
+CREATE INDEX ix_metamorph_transactions_inserted_at_num ON transactions (inserted_at);

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

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

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

@@ -0,0 +1,8 @@
+CREATE TABLE blocks
+(
+    hash         BYTEA PRIMARY KEY,
+    processed_at TIMESTAMPTZ,
+    inserted_at  TIMESTAMPTZ
+);
+
+CREATE INDEX ix_metamorph_blocks_inserted_at ON blocks (inserted_at);