Merge pull request #72 from planetary-social/vanish_subscriber

Add vanish subscriber

cmd/crossposting-service/di/service.go

@@ -24,6 +24,7 @@ type Service struct {
 	migrationsRunner            *migrations.Runner
 	migrations                  migrations.Migrations
 	migrationsProgressCallback  migrations.ProgressCallback
+	vanishSubscriber            *app.VanishSubscriber
 }
 
 func NewService(
@@ -37,6 +38,7 @@ func NewService(
 	migrationsRunner *migrations.Runner,
 	migrations migrations.Migrations,
 	migrationsProgressCallback migrations.ProgressCallback,
+	vanishSubscriber *app.VanishSubscriber,
 ) Service {
 	return Service{
 		app:                         app,
@@ -49,6 +51,7 @@ func NewService(
 		migrationsRunner:            migrationsRunner,
 		migrations:                  migrations,
 		migrationsProgressCallback:  migrationsProgressCallback,
+		vanishSubscriber:            vanishSubscriber,
 	}
 }
 
@@ -97,6 +100,11 @@ func (s Service) Run(ctx context.Context) error {
 		errCh <- errors.Wrap(s.metricsTimer.Run(ctx), "metrics timer error")
 	}()
 
+	runners++
+	go func() {
+		errCh <- errors.Wrap(s.vanishSubscriber.Run(ctx), "vanish subscriver error")
+	}()
+
 	var err error
 	for i := 0; i < runners; i++ {
 		err = multierror.Append(err, errors.Wrap(<-errCh, "error returned by runner"))

cmd/crossposting-service/di/wire.go

@@ -27,6 +27,7 @@ func BuildService(context.Context, config.Config) (Service, func(), error) {
 		applicationSet,
 		sqliteAdaptersSet,
 		downloaderSet,
+		vanishSubscriberSet,
 		memoryPubsubSet,
 		sqlitePubsubSet,
 		loggingSet,
@@ -116,6 +117,10 @@ var downloaderSet = wire.NewSet(
 	app.NewDownloader,
 )
 
+var vanishSubscriberSet = wire.NewSet(
+	app.NewVanishSubscriber,
+)
+
 var tweetGeneratorSet = wire.NewSet(
 	content.NewTransformer,
 	domain.NewTweetGenerator,

compose.yml

@@ -0,0 +1,36 @@
+services:
+  crossposting:
+    platform: linux/amd64
+    build:
+      context: .
+      dockerfile: Dockerfile
+    environment:
+      CROSSPOSTING_TWITTER_KEY: xxx
+      CROSSPOSTING_TWITTER_KEY_SECRET: xxx
+      CROSSPOSTING_DATABASE_PATH: /db/database.sqlite
+      CROSSPOSTING_ENVIRONMENT: DEVELOPMENT
+      CROSSPOSTING_PUBLIC_FACING_ADDRESS: http://localhost:8008/
+      REDIS_URL: redis://redis:6379
+    ports:
+      - 8008:8008
+    volumes:
+      - ./db:/db
+    depends_on:
+      - redis
+
+  redis:
+    platform: linux/amd64
+    image: redis:alpine
+    ports:
+      - 6379:6379
+
+  relay:
+    platform: linux/amd64
+    image: ghcr.io/planetary-social/nosrelay:latest
+    ports:
+      - "7777:7777"
+    environment:
+      - RELAY_URL=wss://example.com
+      - REDIS_URL=redis://redis:6379
+    depends_on:
+      - redis

go.mod

@@ -19,6 +19,7 @@ require (
 	github.com/nbd-wtf/go-nostr v0.25.4
 	github.com/oklog/ulid/v2 v2.1.0
 	github.com/prometheus/client_golang v1.16.0
+	github.com/redis/go-redis/v9 v9.6.2
 	github.com/rs/zerolog v1.29.1
 	github.com/sirupsen/logrus v1.9.3
 	github.com/stretchr/testify v1.8.4
@@ -36,6 +37,7 @@ require (
 	github.com/dghubble/go-twitter v0.0.0-20221104224141-912508c3888b // indirect
 	github.com/dghubble/sling v1.4.1 // indirect
 	github.com/dgraph-io/ristretto v0.1.1 // indirect
+	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 	github.com/dustin/go-humanize v1.0.1 // indirect
 	github.com/gobwas/httphead v0.1.0 // indirect
 	github.com/gobwas/pool v0.2.1 // indirect

go.sum

@@ -7,6 +7,10 @@ github.com/boreq/errors v0.1.0 h1:aJIXv9JnyR5KtxFpQ8/AiblH3nfYmr1e1yoTze/5A1k=
 github.com/boreq/errors v0.1.0/go.mod h1:B3dsXzhYvfgUXp7ViU/moPYM4PojgQ9MiQ21uvY6qqQ=
 github.com/boreq/rest v0.1.0 h1:bAx31Rp1KrXHkCOlzqAtLKdh74xbly2SHkv9k3vX3iA=
 github.com/boreq/rest v0.1.0/go.mod h1:Ckfx0qLDdPbS081820aWkkqvwhlrbv0SDu8UBDY4k7w=
+github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
+github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
+github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
+github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
 github.com/btcsuite/btcd v0.20.1-beta/go.mod h1:wVuoA8VJLEcwgqHBwHmzLRazpKxTv13Px/pDuV7OomQ=
 github.com/btcsuite/btcd v0.22.0-beta.0.20220111032746-97732e52810c/go.mod h1:tjmYdS6MLJ5/s0Fj4DbLgSbDHbEqLJrtnHecBFkdz5M=
 github.com/btcsuite/btcd v0.23.0/go.mod h1:0QJIIN1wwIXF/3G/m87gIwGniDMDQqjVn4SZgnFpsYY=
@@ -61,6 +65,8 @@ github.com/dgraph-io/ristretto v0.1.1 h1:6CWw5tJNgpegArSHpNHJKldNeq03FQCwYvfMVWa
 github.com/dgraph-io/ristretto v0.1.1/go.mod h1:S1GPSBCYCIhmVNfcth17y2zZtQT6wzkzgwUve0VDWWA=
 github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2 h1:tdlZCpZ/P9DhczCTSixgIKmwPv6+wP5DGjqLYw5SUiA=
 github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2/go.mod h1:SqUrOPUnsFjfmXRMNPybcSiG0BgUW2AuFH8PAnS2iTw=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
 github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
 github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
 github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
@@ -171,6 +177,8 @@ github.com/prometheus/procfs v0.10.1 h1:kYK1Va/YMlutzCGazswoHKo//tZVlFpKYh+Pymzi
 github.com/prometheus/procfs v0.10.1/go.mod h1:nwNm2aOCAYw8uTR/9bWRREkZFxAUcWzPHWJq+XBB/FM=
 github.com/puzpuzpuz/xsync/v2 v2.5.1 h1:mVGYAvzDSu52+zaGyNjC+24Xw2bQi3kTr4QJ6N9pIIU=
 github.com/puzpuzpuz/xsync/v2 v2.5.1/go.mod h1:gD2H2krq/w52MfPLE+Uy64TzJDVY7lP2znR9qmR35kU=
+github.com/redis/go-redis/v9 v9.6.2 h1:w0uvkRbc9KpgD98zcvo5IrVUsn0lXpRMuhNgiHDJzdk=
+github.com/redis/go-redis/v9 v9.6.2/go.mod h1:0C0c6ycQsdpVNQpxb1njEQIqkx5UcsM8FJCQLgE9+RA=
 github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
 github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog=
 github.com/rs/xid v1.4.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg=

service/adapters/mocks/public_key_repository.go

@@ -21,6 +21,10 @@ func (m *PublicKeyRepository) Delete(accountID accounts.AccountID, publicKey dom
 	return errors.New("not implemented")
 }
 
+func (m *PublicKeyRepository) DeleteByPublicKey(publicKey domain.PublicKey) error {
+	return errors.New("not implemented")
+}
+
 func (m *PublicKeyRepository) List() ([]*domain.LinkedPublicKey, error) {
 	return nil, errors.New("not implemented")
 }

service/adapters/sqlite/public_key_repository.go

@@ -49,6 +49,27 @@ WHERE account_id = $1 AND public_key = $2
 	return nil
 }
 
+func (m *PublicKeyRepository) DeleteByPublicKey(publicKey domain.PublicKey) error {
+	var accountID string
+	row := m.tx.QueryRow(`
+		SELECT account_id FROM public_keys WHERE public_key = $1
+	`, publicKey.Hex())
+
+	err := row.Scan(&accountID)
+	if err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return errors.New("no account found with the provided public key")
+		}
+		return errors.Wrap(err, "error retrieving account_id")
+	}
+
+	if err := m.deleteAccountData(accountID); err != nil {
+		return errors.Wrap(err, "error deleting account-related data")
+	}
+
+	return nil
+}
+
 func (m *PublicKeyRepository) List() ([]*domain.LinkedPublicKey, error) {
 	rows, err := m.tx.Query(`
 SELECT account_id, public_key, created_at
@@ -139,3 +160,27 @@ func (m *PublicKeyRepository) readPublicKey(row *sql.Rows) (*domain.LinkedPublic
 
 	return domain.NewLinkedPublicKey(accountID, publicKey, createdAt)
 }
+
+func (m *PublicKeyRepository) deleteAccountData(accountID string) error {
+	_, err := m.tx.Exec(`DELETE FROM public_keys WHERE account_id = $1`, accountID)
+	if err != nil {
+		return errors.Wrap(err, "error deleting from public_keys")
+	}
+
+	_, err = m.tx.Exec(`DELETE FROM sessions WHERE account_id = $1`, accountID)
+	if err != nil {
+		return errors.Wrap(err, "error deleting from sessions")
+	}
+
+	_, err = m.tx.Exec(`DELETE FROM accounts WHERE account_id = $1`, accountID)
+	if err != nil {
+		return errors.Wrap(err, "error deleting from accounts")
+	}
+
+	_, err = m.tx.Exec(`DELETE FROM user_tokens WHERE account_id = $1`, accountID)
+	if err != nil {
+		return errors.Wrap(err, "error deleting from user_tokens")
+	}
+
+	return nil
+}

service/adapters/sqlite/public_key_repository_test.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/planetary-social/nos-crossposting-service/internal/fixtures"
 	"github.com/planetary-social/nos-crossposting-service/service/adapters/sqlite"
+	"github.com/planetary-social/nos-crossposting-service/service/app"
 	"github.com/planetary-social/nos-crossposting-service/service/domain"
 	"github.com/planetary-social/nos-crossposting-service/service/domain/accounts"
 	"github.com/stretchr/testify/require"
@@ -259,3 +260,45 @@ func TestPublicKeyRepository_CountCountsPublicKeys(t *testing.T) {
 	})
 	require.NoError(t, err)
 }
+
+func TestPublicKeyRepository_DeleteByPublicKey(t *testing.T) {
+	ctx := fixtures.TestContext(t)
+	adapters := NewTestAdapters(ctx, t)
+
+	accountID := fixtures.SomeAccountID()
+	twitterID := fixtures.SomeTwitterID()
+	publicKey := fixtures.SomePublicKey()
+
+	account, err := accounts.NewAccount(accountID, twitterID)
+	require.NoError(t, err)
+
+	err = adapters.TransactionProvider.Transact(ctx, func(ctx context.Context, adapters sqlite.TestAdapters) error {
+		err = adapters.AccountRepository.Save(account)
+		require.NoError(t, err)
+
+		linkedPublicKey, err := domain.NewLinkedPublicKey(accountID, publicKey, time.Now())
+		require.NoError(t, err)
+		err = adapters.PublicKeyRepository.Save(linkedPublicKey)
+		require.NoError(t, err)
+
+		return nil
+	})
+	require.NoError(t, err)
+
+	err = adapters.TransactionProvider.Transact(ctx, func(ctx context.Context, adapters sqlite.TestAdapters) error {
+		err := adapters.PublicKeyRepository.DeleteByPublicKey(publicKey)
+		require.NoError(t, err)
+
+		return nil
+	})
+	require.NoError(t, err)
+
+	err = adapters.TransactionProvider.Transact(ctx, func(ctx context.Context, adapters sqlite.TestAdapters) error {
+		_, err := adapters.AccountRepository.GetByAccountID(accountID)
+		require.Error(t, err)
+		require.Equal(t, app.ErrAccountDoesNotExist, err)
+
+		return nil
+	})
+	require.NoError(t, err)
+}

service/app/app.go

@@ -49,6 +49,7 @@ type SessionRepository interface {
 type PublicKeyRepository interface {
 	Save(linkedPublicKey *domain.LinkedPublicKey) error
 	Delete(accountID accounts.AccountID, publicKey domain.PublicKey) error
+	DeleteByPublicKey(publicKey domain.PublicKey) error
 	List() ([]*domain.LinkedPublicKey, error)
 	ListByPublicKey(publicKey domain.PublicKey) ([]*domain.LinkedPublicKey, error)
 	ListByAccountID(accountID accounts.AccountID) ([]*domain.LinkedPublicKey, error)