handle token activity

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();

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)
+}

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
+}