diff --git a/database/query/ddl/00024_pn_buy_activity.sql b/database/query/ddl/00024_pn_buy_activity.sql new file mode 100644 index 00000000..d4853984 --- /dev/null +++ b/database/query/ddl/00024_pn_buy_activity.sql @@ -0,0 +1,139 @@ +-- SPDX-License-Identifier: ice License 1.0 + +CREATE TABLE IF NOT EXISTS pn_token_activity ( + id BIGSERIAL PRIMARY KEY, + last_notified_at BIGINT, -- unix timestamp in seconds. + last_processed_at BIGINT, -- unix timestamp in seconds. + window_start_at BIGINT NOT NULL, -- unix timestamp in seconds. + window_end_at BIGINT NOT NULL, -- unix timestamp in seconds. + action_counter BIGINT NOT NULL CHECK (action_counter >= 0), + cfg_time_window INTEGER NOT NULL CHECK (cfg_time_window > 0 AND cfg_time_window <= 31622400), -- seconds. + cfg_count_threshold INTEGER NOT NULL CHECK (cfg_count_threshold > 0), + tc_definition_address TEXT NOT NULL REFERENCES events (address) ON UPDATE CASCADE ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED, + tc_first_action_address TEXT NOT NULL REFERENCES events (address) ON UPDATE CASCADE ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED, + UNIQUE (tc_definition_address) +) WITH (FILLFACTOR = 70); +------ +CREATE OR REPLACE FUNCTION subzero_get_pn_token_activity_config() +RETURNS TABLE ( + cfg_time_window INTEGER, + cfg_count_threshold INTEGER +) +LANGUAGE sql +IMMUTABLE +AS $$ + SELECT + 86400 AS cfg_time_window, + 30 AS cfg_count_threshold; +$$; +------ +CREATE OR REPLACE FUNCTION subzero_get_tx_type(tags jsonb) +RETURNS TEXT +LANGUAGE sql +IMMUTABLE +RETURNS NULL ON NULL INPUT +AS $$ + SELECT + tag->>1 + FROM + jsonb_array_elements(tags) AS tag + WHERE + tag->>0 = 'tx_type' + ORDER BY + tag->>1 + LIMIT 1; +$$; +------ +CREATE OR REPLACE FUNCTION trigger_events_after_insert_track_token_activity() +RETURNS TRIGGER AS $$ +DECLARE + token_address TEXT; +BEGIN + token_address := subzero_get_first_a_tag_value(NEW.tags, 31175); + IF token_address = '' OR token_address IS NULL THEN + RETURN NEW; + END IF; + + MERGE INTO pn_token_activity AS target + USING ( + SELECT + token_address AS tc_definition_address, + COALESCE( + ( + SELECT + e.address + FROM + events e + WHERE + e.kind = 1175 + AND e.hidden = FALSE + AND e.deleted = FALSE + AND subzero_get_first_a_tag_value(e.tags, 31175) = token_address + AND subzero_get_tx_type(e.tags) = 'buy' + ORDER BY + e.lookup_created_at ASC + LIMIT 1 + ), + NEW.address + ) AS tc_first_action_address, + to_timestamp_seconds(NEW.created_at) AS event_created_at, + cfg.cfg_time_window, + cfg.cfg_count_threshold + FROM subzero_get_pn_token_activity_config() cfg + ) AS source + ON (target.tc_definition_address = source.tc_definition_address) + WHEN MATCHED AND source.event_created_at < target.window_start_at THEN + -- Event is outside the current window (too old). + DO NOTHING + WHEN MATCHED + AND (target.last_notified_at IS NULL OR target.last_notified_at < target.window_start_at) + AND (source.event_created_at >= target.window_end_at) + AND (target.action_counter >= target.cfg_count_threshold) + THEN + -- Keep counting while pending, with some buffer after window end to avoid racing with the notification process. + UPDATE SET + window_end_at = (source.event_created_at / 180) * 180, + action_counter = target.action_counter + 1 + WHEN MATCHED AND source.event_created_at >= target.window_end_at THEN + -- Event is outside the current window, and no pending notification. Start a new window. + UPDATE SET + window_start_at = source.event_created_at, + window_end_at = source.event_created_at + target.cfg_time_window, + action_counter = 1 + WHEN MATCHED THEN + -- Event is within the current window, and no pending notification. Just count. + UPDATE SET + action_counter = target.action_counter + 1 + WHEN NOT MATCHED AND EXISTS ( + SELECT 1 + FROM events e + WHERE e.address = source.tc_definition_address + ) THEN + -- New record for a token with activity, but no existing window. Start a new window. + INSERT ( + window_start_at, + window_end_at, + action_counter, + cfg_time_window, + cfg_count_threshold, + tc_definition_address, + tc_first_action_address + ) VALUES ( + source.event_created_at, + source.event_created_at + source.cfg_time_window, + 1, + source.cfg_time_window, + source.cfg_count_threshold, + source.tc_definition_address, + source.tc_first_action_address + ); + + RETURN NEW; +END; +$$ LANGUAGE plpgsql; +------ +CREATE OR REPLACE TRIGGER trigger_events_after_insert_track_token_activity +AFTER INSERT ON events +FOR EACH ROW +WHEN (NEW.kind = 1175 AND NEW.hidden = FALSE AND NEW.deleted = FALSE AND subzero_get_tx_type(NEW.tags) = 'buy') +EXECUTE FUNCTION trigger_events_after_insert_track_token_activity(); diff --git a/database/query/global.go b/database/query/global.go index 9bd5ef55..92225b15 100644 --- a/database/query/global.go +++ b/database/query/global.go @@ -302,3 +302,11 @@ func RegisterPriceChangeSubscriber(ctx context.Context, deviceUUID string, ev *m func CollectPriceChangeSubscribersCandidates(ctx context.Context, ev *model.Event, startID, limit uint64) ([]string, uint64, error) { return globalDB.Client.CollectPriceChangeSubscribersCandidates(ctx, ev, startID, limit) } + +func CollectTokenActivityCandidates(ctx context.Context, now time.Time, startID, limit uint64) ([]string, uint64, error) { + return globalDB.Client.CollectTokenActivityCandidates(ctx, now, startID, limit) +} + +func FetchAndUpdateTokenActivityNotification(ctx context.Context, now time.Time, tokens []string) ([]*TokenActivityData, error) { + return globalDB.Client.FetchAndUpdateTokenActivityNotification(ctx, now, tokens) +} diff --git a/database/query/query_pn_activity.go b/database/query/query_pn_activity.go new file mode 100644 index 00000000..aea96963 --- /dev/null +++ b/database/query/query_pn_activity.go @@ -0,0 +1,151 @@ +// SPDX-License-Identifier: ice License 1.0 + +package query + +import ( + "context" + "time" + + "github.com/cockroachdb/errors" + "github.com/rs/zerolog/log" + + "github.com/ice-blockchain/subzero/database/query/internal/connector" +) + +type ( + TokenActivityData struct { + Definition string + FirstBuy string + Counter uint64 + } +) + +func (client *dbClient) CollectTokenActivityCandidates(ctx context.Context, now time.Time, startID, limit uint64) (defCandidates []string, lastID uint64, err error) { + type record struct { + ID uint64 + Address string + } + const query = ` +WITH to_claim AS ( + SELECT + id + FROM + pn_token_activity + WHERE + window_end_at < :now + AND action_counter >= cfg_count_threshold + AND (last_processed_at IS NULL OR last_processed_at < window_end_at) + AND (last_notified_at IS NULL OR last_notified_at < window_start_at) + AND id > :startID + ORDER BY id ASC + LIMIT :limit + FOR UPDATE SKIP LOCKED +) +UPDATE + pn_token_activity p +SET + last_processed_at = :now +FROM + to_claim c +WHERE + p.id = c.id +RETURNING + p.id, + p.tc_definition_address as address +` + + rows, err := connector.ExecNamed[record](ctx, client.db, query, map[string]any{ + "now": now.Unix(), + "startID": startID, + "limit": limit, + }) + if err != nil { + return nil, startID, errors.Wrap(err, "failed to collect token activity candidates") + } else if len(rows) == 0 { + return nil, startID, nil + } + + defCandidates = make([]string, 0, len(rows)) + lastID = startID + for _, row := range rows { + defCandidates = append(defCandidates, row.Address) + lastID = row.ID + } + + log.Trace(). + Str("context", "DB"). + Time("now", now). + Int("count", len(defCandidates)). + Uint64("last_id", lastID). + Msg("collected token activity candidates") + + return defCandidates, lastID, nil +} + +func (client *dbClient) FetchAndUpdateTokenActivityNotification(ctx context.Context, now time.Time, tokens []string) ([]*TokenActivityData, error) { + const query = ` +WITH current_tokens AS ( + SELECT + id, + tc_definition_address, + tc_first_action_address, + window_start_at, + action_counter + FROM + pn_token_activity + WHERE + tc_definition_address = ANY(:tokens) + AND window_end_at < :now + AND action_counter >= cfg_count_threshold + AND (last_notified_at IS NULL OR (:now-last_notified_at) > cfg_time_window) + FOR UPDATE SKIP LOCKED +), +updated_data AS ( + UPDATE pn_token_activity + SET + last_notified_at = :now, + action_counter = 0 + FROM current_tokens + WHERE + pn_token_activity.id = current_tokens.id + RETURNING + pn_token_activity.id, + pn_token_activity.tc_definition_address, + pn_token_activity.tc_first_action_address, + current_tokens.action_counter +) +SELECT + jsonb_build_object( + 'id', def.id, + 'pubkey', def.pubkey, + 'created_at', def.created_at, + 'kind', def.kind, + 'tags', def.tags, + 'content', def.content, + 'sig', def.sig + ) AS definition, + jsonb_build_object( + 'id', act.id, + 'pubkey', act.pubkey, + 'created_at', act.created_at, + 'kind', act.kind, + 'tags', act.tags, + 'content', act.content, + 'sig', act.sig + ) AS firstbuy, + action_counter as counter +FROM + updated_data ud +INNER JOIN events def ON def.address = ud.tc_definition_address +INNER JOIN events act ON act.address = ud.tc_first_action_address +` + + rows, err := connector.ExecNamed[TokenActivityData](ctx, client.db, query, map[string]any{ + "now": now.Unix(), + "tokens": tokens, + }) + if err != nil { + return nil, errors.Wrap(err, "failed to fetch and update token activity notification") + } + return rows, nil +} diff --git a/database/query/query_pn_activity_test.go b/database/query/query_pn_activity_test.go new file mode 100644 index 00000000..63c7c9c0 --- /dev/null +++ b/database/query/query_pn_activity_test.go @@ -0,0 +1,239 @@ +// SPDX-License-Identifier: ice License 1.0 + +package query + +import ( + "testing" + "time" + + "github.com/nbd-wtf/go-nostr" + "github.com/stretchr/testify/require" + + "github.com/ice-blockchain/subzero/model" +) + +func TestTokenActivityFlow(t *testing.T) { + t.Parallel() + + db := helperNewDatabase(t) + defer db.Close() + + baseTime := nostr.Now().Add(-time.Hour * 48) + + t.Run("No definition event", func(t *testing.T) { + buyEvent := helperNewPriceActionEvent(t, "31175:foo:bar", "10", baseTime.Add(time.Second*100)) + require.NoError(t, db.AcceptEvents(t.Context(), buyEvent)) + + candidates, _, err := db.CollectTokenActivityCandidates(t.Context(), time.Now(), 0, 100) + require.NoError(t, err) + require.Empty(t, candidates) + }) + + t.Run("Regular flow with buy and sell events", func(t *testing.T) { + var post, def model.Event + post.Kind = nostr.KindArticle + post.Content = "test post for token" + post.CreatedAt = baseTime + require.NoError(t, post.SignWithAlg(model.GeneratePrivateKey(), model.SignAlgEDDSA, model.KeyAlgCurve25519)) + + def.Kind = model.CustomIONKindTokenizedCommunityDefinition + def.CreatedAt = baseTime + 1 + def.Content = "token definition" + def.Tags = model.Tags{ + {"a", post.Address()}, + } + require.NoError(t, def.SignWithAlg(model.GeneratePrivateKey(), model.SignAlgEDDSA, model.KeyAlgCurve25519)) + + require.NoError(t, db.AcceptEvents(t.Context(), &post, &def)) + + // Create 35 buy events spread over 1 hour to exceed threshold (30). + var firstBuyEvent *model.Event + const buyCount = 35 + for i := range buyCount { + eventTime := baseTime.Add(time.Hour).Add(time.Second * time.Duration(i)) + buyEvent := helperNewPriceActionEvent(t, def.Address(), "10", eventTime, "buy") + require.NoError(t, db.AcceptEvents(t.Context(), buyEvent)) + if i == 0 { + firstBuyEvent = buyEvent + } + } + + const sellCount = 5 + for i := range sellCount { + eventTime := baseTime.Add(time.Hour * 2).Add(time.Second * time.Duration(i)) + sellEvent := helperNewPriceActionEvent(t, def.Address(), "9", eventTime, "sell") + require.NoError(t, db.AcceptEvents(t.Context(), sellEvent)) // Should be ignored in the trigger. + } + + now := time.Now() + candidates, lastID, err := db.CollectTokenActivityCandidates(t.Context(), now, 0, 100) + require.NoError(t, err) + require.Len(t, candidates, 1) + require.Equal(t, def.Address(), candidates[0]) + require.Greater(t, lastID, uint64(0)) + + data, err := db.FetchAndUpdateTokenActivityNotification(t.Context(), now, candidates) + require.NoError(t, err) + require.Len(t, data, 1) + + notification := data[0] + require.EqualValues(t, buyCount, notification.Counter) + require.NotEmpty(t, notification.Definition) + require.NotEmpty(t, notification.FirstBuy) + + var notifiedFirstBuy, notifiedDef model.Event + require.NoError(t, notifiedFirstBuy.UnmarshalJSON([]byte(notification.FirstBuy))) + require.Equal(t, firstBuyEvent, ¬ifiedFirstBuy) + + require.NoError(t, notifiedDef.UnmarshalJSON([]byte(notification.Definition))) + require.Equal(t, def, notifiedDef) + + candidates2, _, err := db.CollectTokenActivityCandidates(t.Context(), now, 0, 100) + require.NoError(t, err) + require.Empty(t, candidates2) + }) +} + +func TestTokenActivityFlowEdgeCases(t *testing.T) { + t.Parallel() + + const buyCountThreshold = 30 + + db := helperNewDatabase(t) + defer db.Close() + + var post model.Event + post.Kind = nostr.KindArticle + post.Content = "under threshold post" + post.CreatedAt = nostr.Now().Add(-time.Hour * 100) + require.NoError(t, post.SignWithAlg(model.GeneratePrivateKey(), model.SignAlgEDDSA, model.KeyAlgCurve25519)) + require.NoError(t, db.AcceptEvents(t.Context(), &post)) + + t.Run("Under threshold is not collected", func(t *testing.T) { + baseTime := nostr.Now().Add(-time.Hour * 48) + + var def model.Event + def.Kind = model.CustomIONKindTokenizedCommunityDefinition + def.CreatedAt = baseTime + 1 + def.Content = "under threshold def" + def.Tags = model.Tags{{"a", post.Address()}} + require.NoError(t, def.SignWithAlg(model.GeneratePrivateKey(), model.SignAlgEDDSA, model.KeyAlgCurve25519)) + require.NoError(t, db.AcceptEvents(t.Context(), &def)) + + for i := range buyCountThreshold - 1 { + eventTime := baseTime.Add(time.Hour).Add(time.Second * time.Duration(i)) + require.NoError(t, db.AcceptEvents(t.Context(), helperNewPriceActionEvent(t, def.Address(), "10", eventTime, "buy"))) + } + + candidates, _, err := db.CollectTokenActivityCandidates(t.Context(), time.Now(), 0, 100) + require.NoError(t, err) + require.Empty(t, candidates) + }) + + t.Run("Old events are ignored", func(t *testing.T) { + baseTime := nostr.Now().Add(-time.Hour * 48) + + var def model.Event + def.Kind = model.CustomIONKindTokenizedCommunityDefinition + def.CreatedAt = baseTime + 1 + def.Content = "old events def" + def.Tags = model.Tags{{"a", post.Address()}} + require.NoError(t, def.SignWithAlg(model.GeneratePrivateKey(), model.SignAlgEDDSA, model.KeyAlgCurve25519)) + require.NoError(t, db.AcceptEvents(t.Context(), &def)) + + for i := range buyCountThreshold { + eventTime := baseTime.Add(time.Hour).Add(time.Second * time.Duration(i)) + buyEvent := helperNewPriceActionEvent(t, def.Address(), "10", eventTime, "buy") + require.NoError(t, db.AcceptEvents(t.Context(), buyEvent)) + } + + oldBuyEvent := helperNewPriceActionEvent(t, def.Address(), "11", baseTime.Add(time.Minute*30), "buy") + require.NoError(t, db.AcceptEvents(t.Context(), oldBuyEvent)) + + now := time.Now() + candidates, _, err := db.CollectTokenActivityCandidates(t.Context(), now, 0, 100) + require.NoError(t, err) + require.Equal(t, []string{def.Address()}, candidates) + + data, err := db.FetchAndUpdateTokenActivityNotification(t.Context(), now, candidates) + require.NoError(t, err) + require.Len(t, data, 1) + require.EqualValues(t, buyCountThreshold, data[0].Counter) + }) + + t.Run("Window switching after notification", func(t *testing.T) { + baseTime := nostr.Now().Add(-time.Hour * 96) + + var def model.Event + def.Kind = model.CustomIONKindTokenizedCommunityDefinition + def.CreatedAt = baseTime + 1 + def.Content = "window switching def" + def.Tags = model.Tags{{"a", post.Address()}} + require.NoError(t, def.SignWithAlg(model.GeneratePrivateKey(), model.SignAlgEDDSA, model.KeyAlgCurve25519)) + require.NoError(t, db.AcceptEvents(t.Context(), &def)) + + for i := range buyCountThreshold { + eventTime := baseTime.Add(time.Hour).Add(time.Second * time.Duration(i)) + require.NoError(t, db.AcceptEvents(t.Context(), helperNewPriceActionEvent(t, def.Address(), "10", eventTime, "buy"))) + } + + firstNow := baseTime.Add(time.Hour * 26) + candidates, _, err := db.CollectTokenActivityCandidates(t.Context(), firstNow.Time(), 0, 100) + require.NoError(t, err) + require.Equal(t, []string{def.Address()}, candidates) + + data, err := db.FetchAndUpdateTokenActivityNotification(t.Context(), firstNow.Time(), candidates) + require.NoError(t, err) + require.Len(t, data, 1) + require.EqualValues(t, buyCountThreshold, data[0].Counter) + + for i := range buyCountThreshold { + eventTime := baseTime.Add(time.Hour * 28).Add(time.Second * time.Duration(i)) + require.NoError(t, db.AcceptEvents(t.Context(), helperNewPriceActionEvent(t, def.Address(), "12", eventTime, "buy"))) + } + + secondNow := baseTime.Add(time.Hour * 54) + candidates, _, err = db.CollectTokenActivityCandidates(t.Context(), secondNow.Time(), 0, 100) + require.NoError(t, err) + require.Equal(t, []string{def.Address()}, candidates) + + data, err = db.FetchAndUpdateTokenActivityNotification(t.Context(), secondNow.Time(), candidates) + require.NoError(t, err) + require.Len(t, data, 1) + require.EqualValues(t, buyCountThreshold, data[0].Counter) + }) + + t.Run("Pending processing sees overflow event before fetch", func(t *testing.T) { + baseTime := nostr.Now().Add(-time.Hour * 96) + + var def model.Event + def.Kind = model.CustomIONKindTokenizedCommunityDefinition + def.CreatedAt = baseTime + 1 + def.Content = "pending processing def" + def.Tags = model.Tags{{"a", post.Address()}} + require.NoError(t, def.SignWithAlg(model.GeneratePrivateKey(), model.SignAlgEDDSA, model.KeyAlgCurve25519)) + require.NoError(t, db.AcceptEvents(t.Context(), &def)) + + for i := range buyCountThreshold { + eventTime := baseTime.Add(time.Hour).Add(time.Second * time.Duration(i)) + require.NoError(t, db.AcceptEvents(t.Context(), helperNewPriceActionEvent(t, def.Address(), "10", eventTime, "buy"))) + } + + collectNow := baseTime.Add(time.Hour * 26) + candidates, _, err := db.CollectTokenActivityCandidates(t.Context(), collectNow.Time(), 0, 100) + require.NoError(t, err) + require.Equal(t, []string{def.Address()}, candidates) + + pendingEvent := helperNewPriceActionEvent(t, def.Address(), "15", baseTime.Add(time.Hour*25).Add(time.Minute*2).Add(time.Second*5), "buy") + require.NoError(t, db.AcceptEvents(t.Context(), pendingEvent)) + + data, err := db.FetchAndUpdateTokenActivityNotification(t.Context(), collectNow.Time(), candidates) + require.NoError(t, err) + require.Len(t, data, 1) + require.EqualValues(t, buyCountThreshold+1, data[0].Counter) + + candidates, _, err = db.CollectTokenActivityCandidates(t.Context(), collectNow.Time(), 0, 100) + require.NoError(t, err) + require.Empty(t, candidates) + }) +} diff --git a/database/query/query_pn_price_test.go b/database/query/query_pn_price_test.go index db854293..41bc3322 100644 --- a/database/query/query_pn_price_test.go +++ b/database/query/query_pn_price_test.go @@ -16,7 +16,7 @@ import ( "github.com/ice-blockchain/subzero/model" ) -func helperNewPriceActionEvent(t *testing.T, defAddress, amount string, ts nostr.Timestamp) *model.Event { +func helperNewPriceActionEvent(t *testing.T, defAddress, amount string, ts nostr.Timestamp, txType ...string) *model.Event { t.Helper() var action model.Event @@ -28,8 +28,15 @@ func helperNewPriceActionEvent(t *testing.T, defAddress, amount string, ts nostr {"token_symbol", "ICE"}, {"tx_amount", "1", "ICE"}, {"tx_amount", amount, "USD"}, - {"tx_type", "buy"}, } + + if len(txType) > 0 { + require.Contains(t, []string{"buy", "sell"}, txType[0], "txType must be 'buy' or 'sell'") + action.Tags = append(action.Tags, model.Tag{"tx_type", txType[0]}) + } else { + action.Tags = append(action.Tags, model.Tag{"tx_type", "buy"}) + } + require.NoError(t, action.SignWithAlg(model.GeneratePrivateKey(), model.SignAlgEDDSA, model.KeyAlgCurve25519)) return &action diff --git a/dvm/dvm.go b/dvm/dvm.go index 0a8481d0..6cfb27f4 100644 --- a/dvm/dvm.go +++ b/dvm/dvm.go @@ -90,6 +90,12 @@ func mustNewDVM(ctx context.Context, opts ...Option) *dvm { rq.RegisterWorker(server.RQ.Register(), &tokenPriceChangeWorker{ DVM: server, }) + rq.RegisterWorker(server.RQ.Register(), &tokenActivityWorker{ + DVM: server, + }) + server.WG.Go(func() { + server.TokenActivityCollector(ctx) + }) } else { log.Warn(). Str("context", "DVM"). @@ -107,6 +113,57 @@ func mustNewDVM(ctx context.Context, opts ...Option) *dvm { return server } +func (d *dvm) TokenActivityCollector(ctx context.Context) { + fire := make(chan struct{}, 1) + + go func() { + ticker := time.NewTicker(time.Minute) + defer ticker.Stop() + defer close(fire) + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + select { + case fire <- struct{}{}: + default: + log.Debug(). + Str("context", "DVM"). + Msg("skipping token activity collection tick because the previous one is still running") + } + } + } + }() + + log.Trace().Str("context", "DVM").Msg("starting token activity collector") +main: + for range fire { + var startID uint64 + now := time.Now() + for batch := 1; ; batch++ { + candidates, lastID, err := query.CollectTokenActivityCandidates(ctx, now, startID, 100) + if err != nil { + log.Error().Str("context", "DVM").Err(err).Int("batch", batch).Msg("failed to collect token activity candidates") + continue main + } else if len(candidates) == 0 { + continue main + } + + err = d.RQ.Push(ctx, &tokenActivityWorkerArgs{ + BatchNum: batch, + Tokens: candidates, + }) + if err != nil { + log.Error().Str("context", "DVM").Err(err).Int("batch", batch).Msg("failed to push token activity worker for batch") + continue main + } + startID = lastID + } + } +} + func (d *dvm) SubmitResult(ctx context.Context, task *jobInfo, result *model.Event) { d.acceptDVMResponseEvent(result) diff --git a/dvm/token_activity_worker.go b/dvm/token_activity_worker.go new file mode 100644 index 00000000..16464907 --- /dev/null +++ b/dvm/token_activity_worker.go @@ -0,0 +1,96 @@ +// SPDX-License-Identifier: ice License 1.0 + +package dvm + +import ( + "context" + "encoding/json" + "time" + + "github.com/cockroachdb/errors" + "github.com/nbd-wtf/go-nostr" + "github.com/rs/zerolog/log" + + "github.com/ice-blockchain/subzero/database/query" + "github.com/ice-blockchain/subzero/model" + "github.com/ice-blockchain/subzero/rq" +) + +type ( + tokenActivityWorkerArgs struct { + BatchNum int `json:"batch_num"` + Tokens []string `json:"tokens"` + Request *model.Event `json:"request"` + } + tokenActivityWorker struct { + rq.WorkerDefaults[tokenActivityWorkerArgs] + DVM *dvm + } + tokenActivityWorkerJob = rq.Job[tokenActivityWorkerArgs] +) + +func (tokenActivityWorkerArgs) Kind() string { + return "dvm_token_activity_worker_args" +} + +func (w *tokenActivityWorker) Work(ctx context.Context, job *tokenActivityWorkerJob) error { + log.Trace(). + Str("context", "DVM"). + Int("batch_num", job.Args.BatchNum). + Int("num_tokens", len(job.Args.Tokens)). + Msg("starting token activity worker") + + data, err := query.FetchAndUpdateTokenActivityNotification(ctx, time.Now(), job.Args.Tokens) + if err != nil { + return errors.Wrapf(err, "batch %d: failed to fetch and update token activity notification for tokens %v", job.Args.BatchNum, job.Args.Tokens) + } + + for _, entry := range data { + var action, def model.Event + + if err := json.Unmarshal([]byte(entry.Definition), &def); err != nil { + log.Error().Str("context", "DVM").Err(err).Str("definition", entry.Definition).Msg("failed to unmarshal token activity definition event") + continue + } + + if err := json.Unmarshal([]byte(entry.FirstBuy), &action); err != nil { + log.Error().Str("context", "DVM").Err(err).Str("first_buy", entry.FirstBuy).Msg("failed to unmarshal token activity first buy event") + continue + } + + content, err := json.Marshal(model.Events{&action, &def}) + if err != nil { + log.Error().Str("context", "DVM").Err(err).Str("definition", entry.Definition).Str("first_buy", entry.FirstBuy).Msg("failed to marshal events for token activity notification") + continue + } + + var ev model.Event + ev.Kind = model.CustomIONKindDVMJobResponseTrendingTokens + ev.Content = string(content) + ev.CreatedAt = nostr.Now() + ev.Tags = model.Tags{ + {"p", def.GetMasterPublicKey()}, + {"a", def.Address()}, + } + + if job.Args.Request != nil { + ev.Tags = append(ev.Tags, model.Tag{"e", job.Args.Request.ID}) + ev.Tags = append(ev.Tags, model.Tag{"request", job.Args.Request.String()}) + } + + if err := ev.SignWithAlg(w.DVM.Config.PrivateKey, model.SignAlgEDDSA, model.KeyAlgCurve25519); err != nil { + log.Panic().Err(err).Str("context", "DVM").Msg("failed to sign token activity notification event") + } + + log.Trace(). + Str("context", "DVM"). + Int("batch_num", job.Args.BatchNum). + Str("notification_event_id", ev.ID). + Stringer("notification_event", &ev). + Msg("submitting token activity notification event to DVM") + + w.DVM.SubmitResult(ctx, nil, &ev) + } + + return nil +} diff --git a/go.mod b/go.mod index 74cd03a2..63f3281c 100644 --- a/go.mod +++ b/go.mod @@ -12,7 +12,7 @@ replace ( require ( firebase.google.com/go/v4 v4.19.0 - github.com/RoaringBitmap/roaring/v2 v2.15.0 + github.com/RoaringBitmap/roaring/v2 v2.16.0 github.com/alitto/pond/v2 v2.7.0 github.com/bdragon300/tusgo v0.1.3-0.20250807091306-5a1233754936 github.com/bzick/tokenizer v1.4.10 @@ -102,7 +102,7 @@ require ( atomicgo.dev/schedule v0.1.0 // indirect cel.dev/expr v0.25.1 // indirect cloud.google.com/go v0.123.0 // indirect - cloud.google.com/go/auth v0.18.3-0.20260316085740-340a39a83654 // indirect + cloud.google.com/go/auth v0.18.3-0.20260319222451-48c70e2e8670 // indirect cloud.google.com/go/auth/oauth2adapt v0.2.8 // indirect cloud.google.com/go/compute/metadata v0.9.0 // indirect cloud.google.com/go/firestore v1.21.0 // indirect @@ -336,9 +336,9 @@ require ( golang.org/x/time v0.15.0 // indirect golang.org/x/tools v0.43.0 // indirect google.golang.org/appengine/v2 v2.0.6 // indirect - google.golang.org/genproto v0.0.0-20260316180232-0b37fe3546d5 // indirect - google.golang.org/genproto/googleapis/api v0.0.0-20260316180232-0b37fe3546d5 // indirect - google.golang.org/genproto/googleapis/rpc v0.0.0-20260316180232-0b37fe3546d5 // indirect + google.golang.org/genproto v0.0.0-20260319201613-d00831a3d3e7 // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20260319201613-d00831a3d3e7 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260319201613-d00831a3d3e7 // indirect google.golang.org/protobuf v1.36.11 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect gotest.tools/gotestsum v1.13.0 // indirect diff --git a/go.sum b/go.sum index 774fe2a4..7f084b8e 100644 --- a/go.sum +++ b/go.sum @@ -10,8 +10,8 @@ cel.dev/expr v0.25.1 h1:1KrZg61W6TWSxuNZ37Xy49ps13NUovb66QLprthtwi4= cel.dev/expr v0.25.1/go.mod h1:hrXvqGP6G6gyx8UAHSHJ5RGk//1Oj5nXQ2NI02Nrsg4= cloud.google.com/go v0.123.0 h1:2NAUJwPR47q+E35uaJeYoNhuNEM9kM8SjgRgdeOJUSE= cloud.google.com/go v0.123.0/go.mod h1:xBoMV08QcqUGuPW65Qfm1o9Y4zKZBpGS+7bImXLTAZU= -cloud.google.com/go/auth v0.18.3-0.20260316085740-340a39a83654 h1:XkbEraoPkPt1iBVs2Nh/rStKhfTJ+NVb51BE172V7zw= -cloud.google.com/go/auth v0.18.3-0.20260316085740-340a39a83654/go.mod h1:kujE/EN2+Hzs5VdHqLiMXctvft2lCLvy7qpK7pfYLKY= +cloud.google.com/go/auth v0.18.3-0.20260319222451-48c70e2e8670 h1:zvMkXTbWpbyhVWAKgFTjCFxqQ7kOyf7W+KKpGtNwYrI= +cloud.google.com/go/auth v0.18.3-0.20260319222451-48c70e2e8670/go.mod h1:2Aph7BT2KnaSFOM0JDPyiYgNh6PL9vGMiP8CUIXZ+IY= cloud.google.com/go/auth/oauth2adapt v0.2.8 h1:keo8NaayQZ6wimpNSmW5OPc283g65QNIiLpZnkHRbnc= cloud.google.com/go/auth/oauth2adapt v0.2.8/go.mod h1:XQ9y31RkqZCcwJWNSx2Xvric3RrU88hAYYbjDWYDL+c= cloud.google.com/go/compute/metadata v0.9.0 h1:pDUj4QMoPejqq20dK0Pg2N4yG9zIkYGdBtwLoEkH9Zs= @@ -78,8 +78,8 @@ github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERo github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU= github.com/Nvveen/Gotty v0.0.0-20120604004816-cd527374f1e5 h1:TngWCqHvy9oXAN6lEVMRuU21PR1EtLVZJmdB18Gu3Rw= github.com/Nvveen/Gotty v0.0.0-20120604004816-cd527374f1e5/go.mod h1:lmUJ/7eu/Q8D7ML55dXQrVaamCz2vxCfdQBasLZfHKk= -github.com/RoaringBitmap/roaring/v2 v2.15.0 h1:gCbixa3UiG7g6WUZNVOfEEg2HTc1vR4OVdMkX8t1ZFc= -github.com/RoaringBitmap/roaring/v2 v2.15.0/go.mod h1:eq4wdNXxtJIS/oikeCzdX1rBzek7ANzbth041hrU8Q4= +github.com/RoaringBitmap/roaring/v2 v2.16.0 h1:Kys1UNf49d5W8Tq3bpuAhIr/Z8/yPB+59CO8A6c/BbE= +github.com/RoaringBitmap/roaring/v2 v2.16.0/go.mod h1:eq4wdNXxtJIS/oikeCzdX1rBzek7ANzbth041hrU8Q4= github.com/adlio/schema v1.3.6 h1:k1/zc2jNfeiZBA5aFTRy37jlBIuCkXCm0XmvpzCKI9I= github.com/adlio/schema v1.3.6/go.mod h1:qkxwLgPBd1FgLRHYVCmQT/rrBr3JH38J9LjmVzWNudg= github.com/aead/siphash v1.0.1/go.mod h1:Nywa3cDsYNNK3gaciGTWPwHt0wlpNV15vwmswBAUSII= @@ -1029,12 +1029,12 @@ google.golang.org/api v0.272.0 h1:eLUQZGnAS3OHn31URRf9sAmRk3w2JjMx37d2k8AjJmA= google.golang.org/api v0.272.0/go.mod h1:wKjowi5LNJc5qarNvDCvNQBn3rVK8nSy6jg2SwRwzIA= google.golang.org/appengine/v2 v2.0.6 h1:LvPZLGuchSBslPBp+LAhihBeGSiRh1myRoYK4NtuBIw= google.golang.org/appengine/v2 v2.0.6/go.mod h1:WoEXGoXNfa0mLvaH5sV3ZSGXwVmy8yf7Z1JKf3J3wLI= -google.golang.org/genproto v0.0.0-20260316180232-0b37fe3546d5 h1:JNfk58HZ8lfmXbYK2vx/UvsqIL59TzByCxPIX4TDmsE= -google.golang.org/genproto v0.0.0-20260316180232-0b37fe3546d5/go.mod h1:x5julN69+ED4PcFk/XWayw35O0lf/nGa4aNgODCmNmw= -google.golang.org/genproto/googleapis/api v0.0.0-20260316180232-0b37fe3546d5 h1:CogIeEXn4qWYzzQU0QqvYBM8yDF9cFYzDq9ojSpv0Js= -google.golang.org/genproto/googleapis/api v0.0.0-20260316180232-0b37fe3546d5/go.mod h1:EIQZ5bFCfRQDV4MhRle7+OgjNtZ6P1PiZBgAKuxXu/Y= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260316180232-0b37fe3546d5 h1:aJmi6DVGGIStN9Mobk/tZOOQUBbj0BPjZjjnOdoZKts= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260316180232-0b37fe3546d5/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= +google.golang.org/genproto v0.0.0-20260319201613-d00831a3d3e7 h1:XzmzkmB14QhVhgnawEVsOn6OFsnpyxNPRY9QV01dNB0= +google.golang.org/genproto v0.0.0-20260319201613-d00831a3d3e7/go.mod h1:L43LFes82YgSonw6iTXTxXUX1OlULt4AQtkik4ULL/I= +google.golang.org/genproto/googleapis/api v0.0.0-20260319201613-d00831a3d3e7 h1:41r6JMbpzBMen0R/4TZeeAmGXSJC7DftGINUodzTkPI= +google.golang.org/genproto/googleapis/api v0.0.0-20260319201613-d00831a3d3e7/go.mod h1:EIQZ5bFCfRQDV4MhRle7+OgjNtZ6P1PiZBgAKuxXu/Y= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260319201613-d00831a3d3e7 h1:ndE4FoJqsIceKP2oYSnUZqhTdYufCYYkqwtFzfrhI7w= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260319201613-d00831a3d3e7/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= google.golang.org/grpc v1.79.3 h1:sybAEdRIEtvcD68Gx7dmnwjZKlyfuc61Dyo9pGXXkKE= google.golang.org/grpc v1.79.3/go.mod h1:KmT0Kjez+0dde/v2j9vzwoAScgEPx/Bw1CYChhHLrHQ= google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8= diff --git a/model/model.go b/model/model.go index 1a726f1f..ddac5bd9 100644 --- a/model/model.go +++ b/model/model.go @@ -45,6 +45,8 @@ const ( CustomIONKindBlockchainActivityConsent = 4175 CustomIONKindDVMJobRequestPriceChange = 5176 CustomIONKindDVMJobResponsePriceChange = CustomIONKindDVMJobRequestPriceChange + 1000 + CustomIONKindDVMJobRequestTrendingTokens = 5177 + CustomIONKindDVMJobResponseTrendingTokens = CustomIONKindDVMJobRequestTrendingTokens + 1000 CustomIONKindAttestation = 10_100 CustomIONKindRelayListMetadata = 20_002 CustomIONKindEphemeralEmbedding = 21_750 diff --git a/push-notifications/broadcaster_relay_finder.go b/push-notifications/broadcaster_relay_finder.go index ae6a0944..3851e9b2 100644 --- a/push-notifications/broadcaster_relay_finder.go +++ b/push-notifications/broadcaster_relay_finder.go @@ -76,6 +76,11 @@ func (w *broadcasterRelayFinderWorker) Work(ctx context.Context, job *rq.Job[bro masterKeys := make([]string, 0, len(events)) for _, event := range events { keys := w.Manager.collectTargetMasterKeys(event) + if event.IsJobResponse() { + if pTag := event.GetTag("p").Value(); pTag != "" { + keys = append(keys, pTag) + } + } if len(keys) == 0 { continue } diff --git a/push-notifications/notification_dvm.go b/push-notifications/notification_dvm.go index 89d4bf63..a26eee5f 100644 --- a/push-notifications/notification_dvm.go +++ b/push-notifications/notification_dvm.go @@ -26,6 +26,8 @@ func (pm *PushNotificationManager) handleNewDVMEvent(_ context.Context, event *m switch event.Kind { case model.CustomIONKindDVMJobResponsePriceChange: notificationType = NotificationTypeTokenPriceChange + case model.CustomIONKindDVMJobResponseTrendingTokens: + notificationType = NotificationTypeTokenActivity default: log.Warn(). Str("context", "PUSH_NOTIFICATION"). diff --git a/push-notifications/pushnotifications.go b/push-notifications/pushnotifications.go index d1c8b60b..6531d1d2 100644 --- a/push-notifications/pushnotifications.go +++ b/push-notifications/pushnotifications.go @@ -103,6 +103,7 @@ const ( NotificationTypeContentTokenSwapped NotificationType = "content_token_swapped" NotificationTypeTokenPriceChange NotificationType = "token_price_change" + NotificationTypeTokenActivity NotificationType = "token_activity" CompressionMethodZlib = "zlib" ) @@ -242,6 +243,11 @@ var ( Body: "Your token price changed", ImageURL: "https://ice.io/wp-content/uploads/2024/04/ion-logo-2.png", }, + NotificationTypeTokenActivity: { + Title: "Buying activity increased", + Body: "More buyers joined", + ImageURL: "https://ice.io/wp-content/uploads/2024/04/ion-logo-2.png", + }, } allowedPushEventKinds = map[int]struct{}{ nostr.KindArticle: {}, @@ -255,15 +261,17 @@ var ( model.CustomIONKindTokenizedCommunityDefinition: {}, model.CustomIONKindTokenizedCommunityAction: {}, model.CustomIONKindDVMJobResponsePriceChange: {}, + model.CustomIONKindDVMJobResponseTrendingTokens: {}, model.CustomIONKindFundSendNotify: {}, } allowedBroadcastKinds = map[int]struct{}{ model.CustomIONKindTokenizedCommunityAction: {}, model.CustomIONKindTokenizedCommunityDefinition: {}, model.CustomIONKindEditableTextNote: {}, - nostr.KindArticle: {}, - nostr.KindTextNote: {}, - nostr.KindGenericRepost: {}, + model.CustomIONKindDVMJobResponseTrendingTokens: {}, + nostr.KindArticle: {}, + nostr.KindTextNote: {}, + nostr.KindGenericRepost: {}, } globalPushNotificationManager *PushNotificationManager @@ -729,7 +737,7 @@ func (pm *PushNotificationManager) processEvent(ctx context.Context, event *mode case nostr.KindArticle: notifications, err = pm.handleNewPostEvent(ctx, event, relevantEvents...) err = errors.Wrap(err, "failed to handle article event") - case model.CustomIONKindDVMJobResponsePriceChange: + case model.CustomIONKindDVMJobResponsePriceChange, model.CustomIONKindDVMJobResponseTrendingTokens: notifications, err = pm.handleNewDVMEvent(ctx, event, relevantEvents...) err = errors.Wrap(err, "failed to handle DVM job response event") case nostr.KindReaction: