feat: add fdw to link sarafu network db, add cache lookup for transfers inserts

internal/store/pg.go

@@ -36,6 +36,7 @@ type (
 		InsertPoolSwap         string `query:"insert-pool-swap"`
 		InsertPoolDeposit      string `query:"insert-pool-deposit"`
 		InsertPriceQuoteUpdate string `query:"insert-price-quote-update"`
+		CheckAddressExists     string `query:"address-exists"`
 	}
 )
 
@@ -68,6 +69,21 @@ func NewPgStore(o PgOpts) (Store, error) {
 
 func (pg *Pg) InsertTokenTransfer(ctx context.Context, eventPayload event.Event) error {
 	return pg.executeTransaction(ctx, func(tx pgx.Tx) error {
+		var addressExists bool
+
+		if err := tx.QueryRow(
+			ctx,
+			pg.queries.CheckAddressExists,
+			eventPayload.Payload["from"].(string),
+			eventPayload.Payload["to"].(string),
+		).Scan(&addressExists); err != nil {
+			return err
+		}
+
+		if !addressExists {
+			return nil
+		}
+
 		txID, err := pg.insertTx(ctx, tx, eventPayload)
 		if err != nil {
 			return err
@@ -252,6 +268,16 @@ func loadQueries(queriesPath string) (*queries, error) {
 }
 
 func runMigrations(ctx context.Context, dbPool *pgxpool.Pool, migrationsPath string) error {
+	// Safety check
+	// https://github.com/jackc/tern/issues/98
+	envVars := []string{"REMOTE_DB_HOST", "REMOTE_DB_PORT", "REMOTE_DB_NAME", "REMOTE_DB_USER", "REMOTE_DB_PASSWORD"}
+	for _, envVar := range envVars {
+		_, exists := os.LookupEnv(envVar)
+		if !exists {
+			return fmt.Errorf("required env var %s not set", envVar)
+		}
+	}
+
 	const migratorTimeout = 5 * time.Second
 
 	ctx, cancel := context.WithTimeout(ctx, migratorTimeout)