diff --git a/config/scrapers_config.exs b/config/scrapers_config.exs index 297630ad67..e2f9fe2d16 100644 --- a/config/scrapers_config.exs +++ b/config/scrapers_config.exs @@ -12,6 +12,12 @@ config :sanbase, Sanbase.ExternalServices.Coinmarketcap.TickerFetcher, sync_enabled: {:system, "COINMARKETCAP_TICKER_FETCHER_ENABLED", false}, top_projects_to_follow: {:system, "TOP_PROJECTS_TO_FOLLOW", "25"} +config :sanbase, Sanbase.ExternalServices.Coinmarketcap.ProBackfill, + rate_limiter_scale: {:system, "CMC_PRO_BACKFILL_RATE_LIMITER_SCALE_MS", "60000"}, + rate_limiter_limit: {:system, "CMC_PRO_BACKFILL_RATE_LIMITER_LIMIT", "30"}, + rate_limiter_time_between_requests: + {:system, "CMC_PRO_BACKFILL_RATE_LIMITER_TIME_BETWEEN_REQUESTS_MS", "1000"} + config :sanbase, Sanbase.ExternalServices.Etherscan.Requests, apikey: {:system, "ETHERSCAN_APIKEY", ""} @@ -30,7 +36,9 @@ config :sanbase, Oban.Scrapers, cryptocompare_funding_rate_historical_jobs_queue: [limit: 10, paused: true], cryptocompare_funding_rate_historical_jobs_pause_resume_queue: 1, # Twitter queues - twitter_followers_migration_queue: [limit: 25, paused: true] + twitter_followers_migration_queue: [limit: 25, paused: true], + coinmarketcap_pro_backfill_jobs: [limit: 10, paused: false], + coinmarketcap_pro_backfill_control: 2 ], plugins: [ # The default values of interval: 1000, limit: 5000 cause the stager to timeout diff --git a/docs/ops/cmc_pro_backfill_runbook.md b/docs/ops/cmc_pro_backfill_runbook.md new file mode 100644 index 0000000000..cb809c0212 --- /dev/null +++ b/docs/ops/cmc_pro_backfill_runbook.md @@ -0,0 +1,83 @@ +# CMC Pro Backfill Runbook + +## Dry run for one asset + +```elixir +slug = "bitcoin" + +{:ok, precheck} = + Sanbase.ExternalServices.Coinmarketcap.ProBackfill.gap_check( + scope: :single, + slug: slug, + interval: "5m" + ) + +from = precheck.recommended_time_start +to = precheck.recommended_time_end + +{:ok, run} = + Sanbase.ExternalServices.Coinmarketcap.ProBackfill.start_run( + scope: :single, + slug: slug, + time_start: from, + time_end: to, + interval: "5m", + dry_run?: true + ) +``` + +## Verify dry run + +```elixir +Sanbase.ExternalServices.Coinmarketcap.ProBackfill.status(run.id) + +Sanbase.ExternalServices.Coinmarketcap.ProBackfill.gap_check( + scope: :single, + slug: slug, + interval: "5m" +) +``` + +## Actual run for one asset + +```elixir +{:ok, run} = + Sanbase.ExternalServices.Coinmarketcap.ProBackfill.start_run( + scope: :single, + slug: slug, + time_start: from, + time_end: to, + interval: "5m" + ) +``` + +## Actual run for all assets + +```elixir +{:ok, run} = + Sanbase.ExternalServices.Coinmarketcap.ProBackfill.start_run( + scope: :all, + time_start: from, + time_end: to, + interval: "5m" + ) +``` + +## Progress and controls + +```elixir +Sanbase.ExternalServices.Coinmarketcap.ProBackfill.status(run.id) +Sanbase.ExternalServices.Coinmarketcap.ProBackfill.pause_run(run.id) +Sanbase.ExternalServices.Coinmarketcap.ProBackfill.resume_run(run.id) +Sanbase.ExternalServices.Coinmarketcap.ProBackfill.cancel_run(run.id) +Sanbase.ExternalServices.Coinmarketcap.ProBackfill.list_runs(limit: 20) +Sanbase.ExternalServices.Coinmarketcap.ProBackfill.AuditReport.run_report(run.id) +``` + +## Final verification + +Run the same gap check for the completed interval and ensure: + +- `has_gap` is false for fillable ranges +- run status is `completed` +- `failed_assets` is zero or only contains expected deferred ranges diff --git a/lib/sanbase/application/scrapers.ex b/lib/sanbase/application/scrapers.ex index 89dbd2ac56..a6b1f395fa 100644 --- a/lib/sanbase/application/scrapers.ex +++ b/lib/sanbase/application/scrapers.ex @@ -2,6 +2,7 @@ defmodule Sanbase.Application.Scrapers do import Sanbase.ApplicationUtils alias Sanbase.ExternalServices.RateLimiting + alias Sanbase.Utils.Config def init(), do: :ok @@ -38,6 +39,10 @@ defmodule Sanbase.Application.Scrapers do limit: 5, time_between_requests: 2000 ), + RateLimiting.Server.child_spec( + :api_coinmarketcap_backfill_rate_limiter, + backfill_rate_limiter_opts() + ), # Coinmarketcap http rate limiter RateLimiting.Server.child_spec( @@ -100,4 +105,30 @@ defmodule Sanbase.Application.Scrapers do false -> config end end + + defp backfill_rate_limiter_opts() do + [ + scale: + Config.module_get( + Sanbase.ExternalServices.Coinmarketcap.ProBackfill, + :rate_limiter_scale, + "60000" + ) + |> Sanbase.Math.to_integer(), + limit: + Config.module_get( + Sanbase.ExternalServices.Coinmarketcap.ProBackfill, + :rate_limiter_limit, + "30" + ) + |> Sanbase.Math.to_integer(), + time_between_requests: + Config.module_get( + Sanbase.ExternalServices.Coinmarketcap.ProBackfill, + :rate_limiter_time_between_requests, + "1000" + ) + |> Sanbase.Math.to_integer() + ] + end end diff --git a/lib/sanbase/external_services/coinmarketcap/pro_backfill.ex b/lib/sanbase/external_services/coinmarketcap/pro_backfill.ex new file mode 100644 index 0000000000..f8775e334d --- /dev/null +++ b/lib/sanbase/external_services/coinmarketcap/pro_backfill.ex @@ -0,0 +1,238 @@ +defmodule Sanbase.ExternalServices.Coinmarketcap.ProBackfill do + import Ecto.Query + + alias Oban.Job + alias Sanbase.Model.LatestCoinmarketcapData + alias Sanbase.Project + alias Sanbase.Repo + + alias Sanbase.ExternalServices.Coinmarketcap.ProBackfill.{ + Asset, + Run, + RunSeederWorker, + Verification + } + + @oban_conf_name :oban_scrapers + @worker_queue :coinmarketcap_pro_backfill_jobs + @supported_intervals ~w(5m) + + def gap_check(opts), do: Verification.gap_check(opts) + + def start_run(opts) do + with {:ok, interval} <- fetch_interval(Keyword.get(opts, :interval, "5m")), + {:ok, result} <- Verification.gap_check(Keyword.put(opts, :interval, interval)) do + assets_with_gap = + result.assets + |> Enum.filter(&(length(&1.fillable_missing_ranges) > 0)) + + if assets_with_gap == [] do + {:ok, + %{ + status: "no_gap", + total_assets: result.total_assets, + fillable_now_assets: result.fillable_now_assets, + deferred_assets: result.deferred_assets + }} + else + create_run_with_assets(opts, interval, assets_with_gap) + end + end + end + + def pause_run(run_id) do + with %Run{} = run <- Run.get(run_id), + {:ok, _} <- Run.mark_paused(run) do + Oban.pause_queue(@oban_conf_name, queue: @worker_queue) + :ok + else + _ -> {:error, "Run not found"} + end + end + + def resume_run(run_id) do + with %Run{} = run <- Run.get(run_id), + {:ok, _} <- Run.update_run(run, %{status: "running"}) do + Oban.resume_queue(@oban_conf_name, queue: @worker_queue) + :ok + else + _ -> {:error, "Run not found"} + end + end + + def cancel_run(run_id) do + with %Run{} = run <- Run.get(run_id), + {:ok, _} <- Run.mark_canceled(run) do + from(a in Asset, where: a.run_id == ^run_id and a.status in ["pending", "running"]) + |> Repo.update_all(set: [status: "canceled", finished_at: DateTime.utc_now()]) + + :ok + else + _ -> {:error, "Run not found"} + end + end + + def list_runs(opts \\ []) do + Run.list(opts) + |> Enum.map(&run_summary/1) + end + + def status(run_id) do + with %Run{} = run <- Run.get(run_id) do + failed_assets = + Asset.list_failed_by_run(run_id, 10) + |> Enum.map(fn a -> %{slug: a.slug, last_error: a.last_error} end) + + %{ + id: run.id, + status: run.status, + scope: run.scope, + interval: run.interval, + time_start: run.time_start, + time_end: run.time_end, + total_assets: run.total_assets, + done_assets: run.done_assets, + failed_assets: run.failed_assets, + pending_assets: run.pending_assets, + percent_complete: percent_complete(run), + eta_seconds: eta_seconds(run), + running_workers: running_workers(run.id), + top_failed_assets: failed_assets, + api_credits_used_total: run.api_credits_used_total, + api_calls_total: run.api_calls_total, + rate_limited_calls_total: run.rate_limited_calls_total, + usage_precision: run.usage_precision, + dry_run: run.dry_run + } + else + _ -> {:error, "Run not found"} + end + end + + defp create_run_with_assets(opts, interval, assets_with_gap) do + now = DateTime.utc_now() + + attrs = %{ + source: "coinmarketcap", + scope: normalize_scope(opts), + status: "pending", + interval: interval, + time_start: Keyword.fetch!(opts, :time_start), + time_end: Keyword.fetch!(opts, :time_end), + dry_run: Keyword.get(opts, :dry_run?, false), + total_assets: 0, + pending_assets: 0, + started_at: nil, + finished_at: nil + } + + Repo.transaction(fn -> + {:ok, run} = + Run.create(attrs) + + projects_map = + Project.List.projects_with_source("coinmarketcap", + include_hidden: true, + order_by_rank: true + ) + |> Map.new(&{&1.id, &1}) + + rows = + assets_with_gap + |> Enum.map(fn gap -> + project = projects_map[gap.project_id] + cmc_data = LatestCoinmarketcapData.latest_coinmarketcap_data(project) + cmc_integer_id = if(cmc_data, do: cmc_data.coinmarketcap_integer_id, else: nil) + rank = if(cmc_data, do: cmc_data.rank, else: nil) + ranges = %{"ranges" => gap.fillable_missing_ranges} + + {rank || 9_999_999, -(gap.missing_points_count || 0), + %{ + run_id: run.id, + project_id: project.id, + slug: project.slug, + cmc_integer_id: cmc_integer_id, + rank: rank, + status: "pending", + missing_ranges: ranges, + inserted_at: now, + updated_at: now + }} + end) + |> Enum.sort_by( + fn {rank, missing_points_count, _row} -> {rank, missing_points_count} end, + :asc + ) + |> Enum.map(&elem(&1, 2)) + |> Enum.reject(&is_nil(&1.cmc_integer_id)) + + Asset.insert_many(rows) + Run.update_run(run, %{total_assets: length(rows), pending_assets: length(rows)}) + + if rows != [] do + job = RunSeederWorker.new(%{"run_id" => run.id}) + Oban.insert(@oban_conf_name, job) + else + Run.update_run(run, %{status: "completed", finished_at: DateTime.utc_now()}) + end + + run + end) + end + + defp normalize_scope(opts) do + case Keyword.get(opts, :scope) do + scope when scope in [:single, :all, :list] -> Atom.to_string(scope) + scope when scope in ["single", "all", "list"] -> scope + _ -> "all" + end + end + + defp fetch_interval(interval) when interval in @supported_intervals, do: {:ok, interval} + defp fetch_interval(_), do: {:error, "Only 5m interval is currently supported"} + + defp percent_complete(%Run{total_assets: 0}), do: 100.0 + + defp percent_complete(%Run{} = run) do + ((run.done_assets + run.failed_assets) / run.total_assets * 100) |> Float.round(2) + end + + defp eta_seconds(%Run{started_at: nil}), do: nil + defp eta_seconds(%Run{done_assets: 0}), do: nil + + defp eta_seconds(%Run{} = run) do + elapsed = DateTime.diff(DateTime.utc_now(), run.started_at, :second) + avg_per_asset = elapsed / run.done_assets + round(avg_per_asset * run.pending_assets) + end + + defp run_summary(%Run{} = run) do + %{ + id: run.id, + status: run.status, + scope: run.scope, + interval: run.interval, + time_start: run.time_start, + time_end: run.time_end, + total_assets: run.total_assets, + done_assets: run.done_assets, + failed_assets: run.failed_assets, + pending_assets: run.pending_assets, + percent_complete: percent_complete(run), + api_credits_used_total: run.api_credits_used_total, + api_calls_total: run.api_calls_total, + rate_limited_calls_total: run.rate_limited_calls_total, + usage_precision: run.usage_precision + } + end + + defp running_workers(run_id) do + from(j in Job, + where: + j.queue == ^to_string(@worker_queue) and j.state == "executing" and + fragment("?->>'run_id' = ?", j.args, ^to_string(run_id)), + select: count() + ) + |> Repo.one() + end +end diff --git a/lib/sanbase/external_services/coinmarketcap/pro_backfill/asset.ex b/lib/sanbase/external_services/coinmarketcap/pro_backfill/asset.ex new file mode 100644 index 0000000000..9148b71ca6 --- /dev/null +++ b/lib/sanbase/external_services/coinmarketcap/pro_backfill/asset.ex @@ -0,0 +1,128 @@ +defmodule Sanbase.ExternalServices.Coinmarketcap.ProBackfill.Asset do + use Ecto.Schema + + import Ecto.Changeset + import Ecto.Query + + alias Sanbase.Repo + alias __MODULE__ + + @statuses ~w(pending running completed failed canceled) + @usage_precisions ~w(exact estimated mixed) + + schema "coinmarketcap_pro_backfill_assets" do + field(:slug, :string) + field(:cmc_integer_id, :integer) + field(:rank, :integer) + field(:status, :string) + field(:missing_ranges, :map, default: %{"ranges" => []}) + field(:points_exported, :integer, default: 0) + field(:api_credits_used, :float, default: 0.0) + field(:api_calls_total, :integer, default: 0) + field(:rate_limited_calls_total, :integer, default: 0) + field(:usage_precision, :string, default: "exact") + field(:last_error, :string) + field(:started_at, :utc_datetime) + field(:finished_at, :utc_datetime) + + belongs_to(:run, Sanbase.ExternalServices.Coinmarketcap.ProBackfill.Run) + belongs_to(:project, Sanbase.Project) + + timestamps() + end + + def changeset(%Asset{} = asset, attrs) do + asset + |> cast(attrs, [ + :run_id, + :project_id, + :slug, + :cmc_integer_id, + :rank, + :status, + :missing_ranges, + :points_exported, + :api_credits_used, + :api_calls_total, + :rate_limited_calls_total, + :usage_precision, + :last_error, + :started_at, + :finished_at + ]) + |> validate_required([:run_id, :project_id, :slug, :cmc_integer_id, :status]) + |> validate_inclusion(:status, @statuses) + |> validate_inclusion(:usage_precision, @usage_precisions) + |> unique_constraint([:run_id, :project_id]) + end + + def get!(id), do: Repo.get!(Asset, id) + def get(id), do: Repo.get(Asset, id) + + def list_by_run(run_id) do + from(a in Asset, where: a.run_id == ^run_id, order_by: [asc: a.rank, asc: a.project_id]) + |> Repo.all() + end + + def list_pending_by_run(run_id) do + from(a in Asset, + where: a.run_id == ^run_id and a.status == "pending", + order_by: [asc_nulls_last: a.rank, desc: a.points_exported, asc: a.project_id] + ) + |> Repo.all() + end + + def list_failed_by_run(run_id, limit \\ 10) do + from(a in Asset, + where: a.run_id == ^run_id and a.status == "failed", + order_by: [desc: a.updated_at], + limit: ^limit + ) + |> Repo.all() + end + + def update_asset(%Asset{} = asset, attrs) do + asset + |> changeset(attrs) + |> Repo.update() + end + + def mark_running(%Asset{} = asset) do + update_asset(asset, %{status: "running", started_at: asset.started_at || DateTime.utc_now()}) + end + + def mark_completed(%Asset{} = asset, attrs \\ %{}) do + attrs = + Map.merge( + %{ + status: "completed", + finished_at: DateTime.utc_now() + }, + attrs + ) + + update_asset(asset, attrs) + end + + def mark_failed(%Asset{} = asset, error, attrs \\ %{}) do + attrs = + Map.merge( + %{ + status: "failed", + last_error: error, + finished_at: DateTime.utc_now() + }, + attrs + ) + + update_asset(asset, attrs) + end + + def mark_canceled(%Asset{} = asset) do + update_asset(asset, %{status: "canceled", finished_at: DateTime.utc_now()}) + end + + def insert_many(rows) when is_list(rows) do + Repo.insert_all(Asset, rows) + end +end diff --git a/lib/sanbase/external_services/coinmarketcap/pro_backfill/asset_worker.ex b/lib/sanbase/external_services/coinmarketcap/pro_backfill/asset_worker.ex new file mode 100644 index 0000000000..b19f6bca15 --- /dev/null +++ b/lib/sanbase/external_services/coinmarketcap/pro_backfill/asset_worker.ex @@ -0,0 +1,168 @@ +defmodule Sanbase.ExternalServices.Coinmarketcap.ProBackfill.AssetWorker do + use Oban.Worker, + queue: :coinmarketcap_pro_backfill_jobs, + max_attempts: 20, + unique: [period: 60 * 60] + + alias Sanbase.ExternalServices.Coinmarketcap.PricePoint + + alias Sanbase.ExternalServices.Coinmarketcap.ProBackfill.{ + Asset, + ProApiClient, + Run + } + + @prices_exporter :prices_exporter + + @impl Oban.Worker + def perform(%Oban.Job{args: %{"run_id" => run_id, "asset_id" => asset_id}}) do + with %Run{} = run <- Run.get(run_id), + %Asset{} = asset <- Asset.get(asset_id) do + execute(run, asset) + else + _ -> :ok + end + end + + defp execute(%Run{status: "paused"}, _asset), do: {:snooze, 30} + + defp execute(%Run{status: status}, %Asset{} = asset) + when status in ["canceled", "completed", "failed"] do + Asset.mark_canceled(asset) + :ok + end + + defp execute(%Run{}, %Asset{status: status}) + when status in ["completed", "failed", "canceled"], + do: :ok + + defp execute(%Run{} = run, %Asset{} = asset) do + with {:ok, _asset} <- Asset.mark_running(asset), + {:ok, result} <- fetch_all_ranges(run, asset), + {:ok, _} <- maybe_export(run, asset.slug, result.price_points) do + points_exported = length(result.price_points) + + Asset.mark_completed(asset, %{ + points_exported: points_exported, + api_credits_used: result.usage.api_credits_used, + api_calls_total: result.usage.api_calls_total, + rate_limited_calls_total: result.usage.rate_limited_calls_total, + usage_precision: result.usage.usage_precision + }) + + Run.increment_stats(run.id, %{ + done_assets: 1, + pending_assets: -1, + api_credits_used_total: result.usage.api_credits_used, + api_calls_total: result.usage.api_calls_total, + rate_limited_calls_total: result.usage.rate_limited_calls_total, + usage_precision: result.usage.usage_precision + }) + + Run.get(run.id) + |> Run.maybe_mark_completed() + + :ok + else + {:snooze, seconds, usage} -> + Run.increment_stats(run.id, %{ + api_calls_total: usage.api_calls_total || 1, + rate_limited_calls_total: usage.rate_limited_calls_total || 1 + }) + + Asset.update_asset(asset, %{ + api_calls_total: asset.api_calls_total + (usage.api_calls_total || 1), + rate_limited_calls_total: + asset.rate_limited_calls_total + (usage.rate_limited_calls_total || 1) + }) + + {:snooze, seconds} + + {:error, error} -> + Asset.mark_failed(asset, error) + + Run.increment_stats(run.id, %{ + failed_assets: 1, + pending_assets: -1, + last_error: error + }) + + Run.get(run.id) + |> Run.maybe_mark_completed() + + {:error, error} + end + end + + defp maybe_export(%Run{dry_run: true}, _slug, _price_points), do: {:ok, :dry_run} + + defp maybe_export(%Run{}, slug, price_points) do + price_points + |> PricePoint.sanity_filters(slug) + |> Enum.map(&PricePoint.json_kv_tuple(&1, slug)) + |> Sanbase.KafkaExporter.persist_sync(@prices_exporter) + + {:ok, :exported} + rescue + e -> + {:error, Exception.message(e)} + end + + defp fetch_all_ranges(run, asset) do + ranges = + asset.missing_ranges + |> Map.get("ranges", []) + |> Enum.map(&normalize_range/1) + + Enum.reduce_while(ranges, {:ok, %{price_points: [], usage: usage_zero()}}, fn range, + {:ok, acc} -> + case ProApiClient.fetch_range(asset.cmc_integer_id, range.from_unix, range.to_unix, + interval: run.interval + ) do + {:ok, points, usage} -> + merged = %{ + price_points: acc.price_points ++ points, + usage: usage_add(acc.usage, usage) + } + + {:cont, {:ok, merged}} + + {:rate_limited, seconds, usage} -> + {:halt, {:snooze, seconds, usage}} + + {:error, error} -> + {:halt, {:error, error}} + end + end) + end + + defp normalize_range(%{"from_unix" => from_unix, "to_unix" => to_unix}), + do: %{from_unix: from_unix, to_unix: to_unix} + + defp normalize_range(%{from_unix: from_unix, to_unix: to_unix}), + do: %{from_unix: from_unix, to_unix: to_unix} + + defp usage_zero do + %{ + api_credits_used: 0.0, + api_calls_total: 0, + rate_limited_calls_total: 0, + usage_precision: "exact" + } + end + + defp usage_add(left, right) do + %{ + api_credits_used: (left.api_credits_used || 0.0) + (right[:api_credits_used] || 0.0), + api_calls_total: (left.api_calls_total || 0) + (right[:api_calls_total] || 0), + rate_limited_calls_total: + (left.rate_limited_calls_total || 0) + (right[:rate_limited_calls_total] || 0), + usage_precision: usage_precision(left.usage_precision, right[:usage_precision] || "exact") + } + end + + defp usage_precision("mixed", _), do: "mixed" + defp usage_precision(_, "mixed"), do: "mixed" + defp usage_precision(a, a), do: a + defp usage_precision(_, _), do: "mixed" +end diff --git a/lib/sanbase/external_services/coinmarketcap/pro_backfill/audit_report.ex b/lib/sanbase/external_services/coinmarketcap/pro_backfill/audit_report.ex new file mode 100644 index 0000000000..0b9d932433 --- /dev/null +++ b/lib/sanbase/external_services/coinmarketcap/pro_backfill/audit_report.ex @@ -0,0 +1,32 @@ +defmodule Sanbase.ExternalServices.Coinmarketcap.ProBackfill.AuditReport do + alias Sanbase.ExternalServices.Coinmarketcap.ProBackfill + + def run_report(run_id) do + case ProBackfill.status(run_id) do + {:error, _} = error -> + error + + status -> + summary = %{ + run_id: status.id, + status: status.status, + percent_complete: status.percent_complete, + counts: %{ + total_assets: status.total_assets, + done_assets: status.done_assets, + failed_assets: status.failed_assets, + pending_assets: status.pending_assets + }, + api_usage: %{ + api_credits_used_total: status.api_credits_used_total, + api_calls_total: status.api_calls_total, + rate_limited_calls_total: status.rate_limited_calls_total, + usage_precision: status.usage_precision + }, + top_failed_assets: status.top_failed_assets + } + + {:ok, summary} + end + end +end diff --git a/lib/sanbase/external_services/coinmarketcap/pro_backfill/pro_api_client.ex b/lib/sanbase/external_services/coinmarketcap/pro_backfill/pro_api_client.ex new file mode 100644 index 0000000000..eabb786def --- /dev/null +++ b/lib/sanbase/external_services/coinmarketcap/pro_backfill/pro_api_client.ex @@ -0,0 +1,100 @@ +defmodule Sanbase.ExternalServices.Coinmarketcap.ProBackfill.ProApiClient do + alias Sanbase.ExternalServices.Coinmarketcap.PricePoint + alias Sanbase.ExternalServices.RateLimiting.Server + alias Sanbase.Utils.Config + + @rate_limiting_server :api_coinmarketcap_backfill_rate_limiter + @path "/v2/cryptocurrency/quotes/historical" + + def fetch_range(cmc_integer_id, from_unix, to_unix, opts \\ []) do + params = %{ + id: cmc_integer_id, + interval: Keyword.get(opts, :interval, "5m"), + convert: Keyword.get(opts, :convert, "USD,BTC"), + time_start: DateTime.from_unix!(from_unix) |> DateTime.to_iso8601(), + time_end: DateTime.from_unix!(to_unix) |> DateTime.to_iso8601() + } + + Server.wait(@rate_limiting_server) + + case Req.get(base_url: base_url(), url: @path, headers: headers(), params: params) do + {:ok, %{status: 200, body: body}} -> + with {:ok, price_points} <- body_to_price_points(body) do + {:ok, price_points, usage_from_body(body, 1, 0)} + end + + {:ok, %{status: 429, headers: headers}} -> + wait_seconds = header_value(headers, "retry-after") |> Sanbase.Math.to_integer() |> max(1) + wait_until = Timex.shift(Timex.now(), seconds: wait_seconds) + Server.wait_until(@rate_limiting_server, wait_until) + {:rate_limited, wait_seconds, %{api_calls_total: 1, rate_limited_calls_total: 1}} + + {:ok, %{status: status, body: body}} -> + {:error, "CoinMarketCap Pro API status #{status}. Body: #{inspect(body)}"} + + {:error, error} -> + {:error, inspect(error)} + end + end + + defp body_to_price_points(%{"status" => %{"error_code" => 0}, "data" => data}) do + quotes = Map.get(data, "quotes", []) + + price_points = + quotes + |> Enum.map(fn %{"timestamp" => timestamp, "quote" => quote} -> + usd = Map.get(quote, "USD", %{}) + btc = Map.get(quote, "BTC", %{}) + + %PricePoint{ + datetime: Sanbase.DateTimeUtils.from_iso8601!(timestamp), + price_usd: Sanbase.Math.to_float(Map.get(usd, "price")), + price_btc: Sanbase.Math.to_float(Map.get(btc, "price")), + marketcap_usd: Sanbase.Math.to_integer(Map.get(usd, "market_cap")), + volume_usd: Sanbase.Math.to_integer(Map.get(usd, "volume_24h")) + } + end) + + {:ok, price_points} + end + + defp body_to_price_points(body) do + {:error, "Unexpected CMC Pro response: #{inspect(body)}"} + end + + defp usage_from_body(%{"status" => status}, api_calls, rate_limited_calls) do + credits = Map.get(status, "credit_count", 0) |> Sanbase.Math.to_float() + + %{ + api_credits_used: credits, + api_calls_total: api_calls, + rate_limited_calls_total: rate_limited_calls, + usage_precision: "exact" + } + end + + defp usage_from_body(_body, api_calls, rate_limited_calls) do + %{ + api_credits_used: 0.0, + api_calls_total: api_calls, + rate_limited_calls_total: rate_limited_calls, + usage_precision: "estimated" + } + end + + defp headers do + [ + {"X-CMC_PRO_API_KEY", Config.module_get(Sanbase.ExternalServices.Coinmarketcap, :api_key)}, + {"Accept", "application/json"} + ] + end + + defp base_url do + Config.module_get(Sanbase.ExternalServices.Coinmarketcap, :api_url) + end + + defp header_value(headers, key) do + headers + |> Enum.find_value(fn {k, v} -> if String.downcase(k) == key, do: v, else: nil end) + end +end diff --git a/lib/sanbase/external_services/coinmarketcap/pro_backfill/run.ex b/lib/sanbase/external_services/coinmarketcap/pro_backfill/run.ex new file mode 100644 index 0000000000..18a3fe9966 --- /dev/null +++ b/lib/sanbase/external_services/coinmarketcap/pro_backfill/run.ex @@ -0,0 +1,147 @@ +defmodule Sanbase.ExternalServices.Coinmarketcap.ProBackfill.Run do + use Ecto.Schema + + import Ecto.Changeset + import Ecto.Query + + alias Sanbase.Repo + alias __MODULE__ + + @statuses ~w(pending running paused completed failed canceled) + @scopes ~w(single all list) + @usage_precisions ~w(exact estimated mixed) + + schema "coinmarketcap_pro_backfill_runs" do + field(:source, :string) + field(:scope, :string) + field(:status, :string) + field(:interval, :string) + field(:time_start, :utc_datetime) + field(:time_end, :utc_datetime) + field(:dry_run, :boolean, default: false) + field(:total_assets, :integer, default: 0) + field(:done_assets, :integer, default: 0) + field(:failed_assets, :integer, default: 0) + field(:pending_assets, :integer, default: 0) + field(:api_credits_used_total, :float, default: 0.0) + field(:api_calls_total, :integer, default: 0) + field(:rate_limited_calls_total, :integer, default: 0) + field(:usage_precision, :string, default: "exact") + field(:last_error, :string) + field(:started_at, :utc_datetime) + field(:finished_at, :utc_datetime) + + has_many(:assets, Sanbase.ExternalServices.Coinmarketcap.ProBackfill.Asset, + foreign_key: :run_id + ) + + timestamps() + end + + def changeset(%Run{} = run, attrs) do + run + |> cast(attrs, [ + :source, + :scope, + :status, + :interval, + :time_start, + :time_end, + :dry_run, + :total_assets, + :done_assets, + :failed_assets, + :pending_assets, + :api_credits_used_total, + :api_calls_total, + :rate_limited_calls_total, + :usage_precision, + :last_error, + :started_at, + :finished_at + ]) + |> validate_required([:scope, :status, :interval, :time_start, :time_end, :source]) + |> validate_inclusion(:status, @statuses) + |> validate_inclusion(:scope, @scopes) + |> validate_inclusion(:usage_precision, @usage_precisions) + end + + def create(attrs) do + %Run{} + |> changeset(attrs) + |> Repo.insert() + end + + def get(id), do: Repo.get(Run, id) + + def get!(id), do: Repo.get!(Run, id) + + def list(opts \\ []) do + limit = Keyword.get(opts, :limit, 20) + + from(r in Run, order_by: [desc: r.inserted_at], limit: ^limit) + |> Repo.all() + end + + def update_run(%Run{} = run, attrs) do + run + |> changeset(attrs) + |> Repo.update() + end + + def mark_running(%Run{} = run) do + attrs = %{status: "running", started_at: run.started_at || DateTime.utc_now()} + update_run(run, attrs) + end + + def mark_paused(%Run{} = run), do: update_run(run, %{status: "paused"}) + + def mark_canceled(%Run{} = run), + do: update_run(run, %{status: "canceled", finished_at: DateTime.utc_now()}) + + def mark_failed(%Run{} = run, error) do + update_run(run, %{status: "failed", last_error: error, finished_at: DateTime.utc_now()}) + end + + def maybe_mark_completed(%Run{} = run) do + if run.total_assets > 0 and run.done_assets + run.failed_assets >= run.total_assets do + update_run(run, %{status: "completed", finished_at: DateTime.utc_now()}) + else + {:ok, run} + end + end + + def increment_stats(run_id, stats) when is_map(stats) do + query = from(r in Run, where: r.id == ^run_id) + + inc_values = + stats + |> Map.take([ + :done_assets, + :failed_assets, + :pending_assets, + :api_credits_used_total, + :api_calls_total, + :rate_limited_calls_total + ]) + |> Enum.reject(fn {_k, v} -> v == 0 end) + + set_values = + stats + |> Map.take([:usage_precision, :last_error, :status, :finished_at, :started_at]) + |> Enum.reject(fn {_k, v} -> is_nil(v) end) + + opts = + [] + |> maybe_put_update_opt(:inc, inc_values) + |> maybe_put_update_opt(:set, set_values) + + case opts do + [] -> {0, nil} + _ -> Repo.update_all(query, opts) + end + end + + defp maybe_put_update_opt(opts, _key, []), do: opts + defp maybe_put_update_opt(opts, key, values), do: Keyword.put(opts, key, values) +end diff --git a/lib/sanbase/external_services/coinmarketcap/pro_backfill/run_seeder_worker.ex b/lib/sanbase/external_services/coinmarketcap/pro_backfill/run_seeder_worker.ex new file mode 100644 index 0000000000..e5f1fc0503 --- /dev/null +++ b/lib/sanbase/external_services/coinmarketcap/pro_backfill/run_seeder_worker.ex @@ -0,0 +1,42 @@ +defmodule Sanbase.ExternalServices.Coinmarketcap.ProBackfill.RunSeederWorker do + use Oban.Worker, + queue: :coinmarketcap_pro_backfill_control, + max_attempts: 10, + unique: [period: 60 * 60] + + alias Sanbase.ExternalServices.Coinmarketcap.ProBackfill.{Asset, AssetWorker, Run} + + @oban_conf_name :oban_scrapers + + @impl Oban.Worker + def perform(%Oban.Job{args: %{"run_id" => run_id}}) do + with %Run{} = run <- Run.get(run_id), + {:ok, run} <- Run.mark_running(run) do + if run.status == "canceled" do + :ok + else + assets = Asset.list_pending_by_run(run.id) + + jobs = + Enum.map(assets, fn asset -> + AssetWorker.new(%{ + run_id: run.id, + asset_id: asset.id + }) + end) + + case jobs do + [] -> + Run.maybe_mark_completed(run) + :ok + + _ -> + Oban.insert_all(@oban_conf_name, jobs) + :ok + end + end + else + _ -> :ok + end + end +end diff --git a/lib/sanbase/external_services/coinmarketcap/pro_backfill/verification.ex b/lib/sanbase/external_services/coinmarketcap/pro_backfill/verification.ex new file mode 100644 index 0000000000..65a49a18e6 --- /dev/null +++ b/lib/sanbase/external_services/coinmarketcap/pro_backfill/verification.ex @@ -0,0 +1,212 @@ +defmodule Sanbase.ExternalServices.Coinmarketcap.ProBackfill.Verification do + alias Sanbase.Price + alias Sanbase.Project + + @step_seconds 300 + @max_fillable_interval "5m" + @default_discovery_days 7 + + def gap_check(opts) do + with {:ok, scope} <- fetch_scope(opts), + {:ok, from, to} <- fetch_interval(opts, scope) do + projects = projects_for_scope(scope, opts) + + per_asset = + projects + |> Enum.map(&gap_check_project(&1, from, to)) + + missing_assets = + per_asset + |> Enum.filter(&(length(&1.fillable_missing_ranges) > 0)) + + fillable_now_assets = Enum.count(per_asset, &(length(&1.fillable_missing_ranges) > 0)) + deferred_assets = Enum.count(per_asset, &(length(&1.deferred_missing_ranges) > 0)) + {recommended_from, recommended_to} = recommended_interval(per_asset) + + {:ok, + %{ + scope: scope, + interval: @max_fillable_interval, + time_start: from, + time_end: to, + has_gap: missing_assets != [], + total_assets: length(per_asset), + fillable_now_assets: fillable_now_assets, + deferred_assets: deferred_assets, + recommended_time_start: recommended_from, + recommended_time_end: recommended_to, + assets: per_asset + }} + end + end + + def gap_check_project(%Project{slug: slug} = project, %DateTime{} = from, %DateTime{} = to) do + expected = + expected_timestamps(from, to) + |> MapSet.new() + + actual = + case Price.timeseries_metric_data(slug, "price_usd", from, to, @max_fillable_interval, + source: "coinmarketcap" + ) do + {:ok, points} -> + points + |> Enum.map(&DateTime.to_unix(&1.datetime)) + |> MapSet.new() + + _ -> + MapSet.new() + end + + missing = MapSet.difference(expected, actual) |> MapSet.to_list() |> Enum.sort() + missing_ranges = timestamps_to_ranges(missing) + {fillable_ranges, deferred_ranges} = split_fillable_ranges(missing_ranges) + + %{ + project_id: project.id, + slug: slug, + fillable_missing_ranges: fillable_ranges, + deferred_missing_ranges: deferred_ranges, + missing_points_count: length(missing), + expected_points_count: MapSet.size(expected), + actual_points_count: MapSet.size(actual) + } + end + + def expected_timestamps(%DateTime{} = from, %DateTime{} = to) do + from_unix = DateTime.to_unix(from) + to_unix = DateTime.to_unix(to) + + Stream.unfold(from_unix, fn ts -> + if ts <= to_unix do + {ts, ts + @step_seconds} + else + nil + end + end) + |> Enum.to_list() + end + + def timestamps_to_ranges([]), do: [] + + def timestamps_to_ranges([first | rest]) do + {ranges, start_ts, end_ts} = + Enum.reduce(rest, {[], first, first}, fn ts, {ranges, start_ts, end_ts} -> + if ts == end_ts + @step_seconds do + {ranges, start_ts, ts} + else + {[%{from_unix: start_ts, to_unix: end_ts} | ranges], ts, ts} + end + end) + + Enum.reverse([%{from_unix: start_ts, to_unix: end_ts} | ranges]) + end + + def split_fillable_ranges(ranges) do + fillable_to_unix = fillable_to_unix() + + Enum.reduce(ranges, {[], []}, fn %{from_unix: from_unix, to_unix: to_unix}, + {fillable, deferred} -> + cond do + from_unix > fillable_to_unix -> + {fillable, [%{from_unix: from_unix, to_unix: to_unix} | deferred]} + + to_unix <= fillable_to_unix -> + {[%{from_unix: from_unix, to_unix: to_unix} | fillable], deferred} + + true -> + split_fillable = %{from_unix: from_unix, to_unix: fillable_to_unix} + split_deferred = %{from_unix: fillable_to_unix + @step_seconds, to_unix: to_unix} + {[split_fillable | fillable], [split_deferred | deferred]} + end + end) + |> then(fn {fillable, deferred} -> + { + Enum.reverse(fillable) |> Enum.filter(&range_valid?/1), + Enum.reverse(deferred) |> Enum.filter(&range_valid?/1) + } + end) + end + + defp fetch_scope(opts) do + case Keyword.get(opts, :scope) do + scope when scope in [:single, :all, :list] -> {:ok, scope} + scope when scope in ["single", "all", "list"] -> {:ok, String.to_existing_atom(scope)} + _ -> {:error, "Invalid scope. Expected :single, :all or :list"} + end + rescue + ArgumentError -> {:error, "Invalid scope. Expected :single, :all or :list"} + end + + defp fetch_interval(opts, scope) do + from = Keyword.get(opts, :time_start) + to = Keyword.get(opts, :time_end) + + case {from, to} do + {%DateTime{} = from, %DateTime{} = to} -> + {:ok, from, to} + + {nil, nil} when scope == :single -> + discovery_to_unix = fillable_to_unix() + discovery_from_unix = discovery_to_unix - @default_discovery_days * 86_400 + + {:ok, DateTime.from_unix!(discovery_from_unix), DateTime.from_unix!(discovery_to_unix)} + + {nil, nil} -> + {:error, "Both :time_start and :time_end are required for :all and :list scopes"} + + _ -> + {:error, "Both :time_start and :time_end must be DateTime"} + end + end + + defp projects_for_scope(:all, _opts) do + Project.List.projects_with_source("coinmarketcap", include_hidden: true, order_by_rank: true) + end + + defp projects_for_scope(:single, opts) do + case Keyword.get(opts, :slug) do + slug when is_binary(slug) -> + case Project.by_slug(slug) do + %Project{} = project -> [project] + nil -> [] + end + + _ -> + [] + end + end + + defp projects_for_scope(:list, opts) do + slugs = Keyword.get(opts, :slugs, []) |> List.wrap() + + slugs + |> Enum.map(&Project.by_slug/1) + |> Enum.reject(&is_nil/1) + end + + defp fillable_to_unix do + Date.utc_today() + |> DateTime.new!(~T[00:00:00], "Etc/UTC") + |> DateTime.to_unix() + |> Kernel.-(@step_seconds) + end + + defp recommended_interval(per_asset) do + all_fillable_ranges = + per_asset + |> Enum.flat_map(& &1.fillable_missing_ranges) + + case all_fillable_ranges do + [] -> + {nil, nil} + + ranges -> + from_unix = ranges |> Enum.map(& &1.from_unix) |> Enum.min() + to_unix = ranges |> Enum.map(& &1.to_unix) |> Enum.max() + {DateTime.from_unix!(from_unix), DateTime.from_unix!(to_unix)} + end + end + + defp range_valid?(%{from_unix: from_unix, to_unix: to_unix}), do: from_unix <= to_unix +end diff --git a/priv/repo/migrations/20260306173000_create_coinmarketcap_pro_backfill_tables.exs b/priv/repo/migrations/20260306173000_create_coinmarketcap_pro_backfill_tables.exs new file mode 100644 index 0000000000..1db40d53a4 --- /dev/null +++ b/priv/repo/migrations/20260306173000_create_coinmarketcap_pro_backfill_tables.exs @@ -0,0 +1,59 @@ +defmodule Sanbase.Repo.Migrations.CreateCoinmarketcapProBackfillTables do + use Ecto.Migration + + def change do + create table(:coinmarketcap_pro_backfill_runs) do + add(:source, :string, null: false, default: "coinmarketcap") + add(:scope, :string, null: false) + add(:status, :string, null: false, default: "pending") + add(:interval, :string, null: false, default: "5m") + add(:time_start, :utc_datetime, null: false) + add(:time_end, :utc_datetime, null: false) + add(:dry_run, :boolean, null: false, default: false) + add(:total_assets, :integer, null: false, default: 0) + add(:done_assets, :integer, null: false, default: 0) + add(:failed_assets, :integer, null: false, default: 0) + add(:pending_assets, :integer, null: false, default: 0) + add(:api_credits_used_total, :float, null: false, default: 0.0) + add(:api_calls_total, :integer, null: false, default: 0) + add(:rate_limited_calls_total, :integer, null: false, default: 0) + add(:usage_precision, :string, null: false, default: "exact") + add(:last_error, :text) + add(:started_at, :utc_datetime) + add(:finished_at, :utc_datetime) + + timestamps() + end + + create(index(:coinmarketcap_pro_backfill_runs, [:status])) + create(index(:coinmarketcap_pro_backfill_runs, [:inserted_at])) + + create table(:coinmarketcap_pro_backfill_assets) do + add(:run_id, references(:coinmarketcap_pro_backfill_runs, on_delete: :delete_all), + null: false + ) + + add(:project_id, references(:project, on_delete: :delete_all), null: false) + add(:slug, :string, null: false) + add(:cmc_integer_id, :integer, null: false) + add(:rank, :integer) + add(:status, :string, null: false, default: "pending") + add(:missing_ranges, :map, null: false, default: %{}) + add(:points_exported, :integer, null: false, default: 0) + add(:api_credits_used, :float, null: false, default: 0.0) + add(:api_calls_total, :integer, null: false, default: 0) + add(:rate_limited_calls_total, :integer, null: false, default: 0) + add(:usage_precision, :string, null: false, default: "exact") + add(:last_error, :text) + add(:started_at, :utc_datetime) + add(:finished_at, :utc_datetime) + + timestamps() + end + + create(unique_index(:coinmarketcap_pro_backfill_assets, [:run_id, :project_id])) + create(index(:coinmarketcap_pro_backfill_assets, [:run_id, :status])) + create(index(:coinmarketcap_pro_backfill_assets, [:run_id, :rank])) + create(index(:coinmarketcap_pro_backfill_assets, [:slug])) + end +end diff --git a/priv/repo/migrations/20260306190000_add_project_fk_to_coinmarketcap_pro_backfill_assets.exs b/priv/repo/migrations/20260306190000_add_project_fk_to_coinmarketcap_pro_backfill_assets.exs new file mode 100644 index 0000000000..0dcc91daf6 --- /dev/null +++ b/priv/repo/migrations/20260306190000_add_project_fk_to_coinmarketcap_pro_backfill_assets.exs @@ -0,0 +1,7 @@ +defmodule Sanbase.Repo.Migrations.AddProjectFkToCoinmarketcapProBackfillAssets do + use Ecto.Migration + + def change do + :ok + end +end diff --git a/priv/repo/structure.sql b/priv/repo/structure.sql index e2c6510009..5129fb545d 100644 --- a/priv/repo/structure.sql +++ b/priv/repo/structure.sql @@ -2,10 +2,10 @@ -- PostgreSQL database dump -- -\restrict lPrBCVGzfDAQrezJuUuvsvvIxIMIDCrQVbcCW2A0GafidncDVtkqtUHNwYPqA5k +\restrict rEumk3or3c8G96Hxf37AhoShjhR5C9TGx5myvsAuvbIk1ywnIRepGcrCCCkez28 --- Dumped from database version 15.16 (Homebrew) --- Dumped by pg_dump version 15.16 (Homebrew) +-- Dumped from database version 15.15 (Homebrew) +-- Dumped by pg_dump version 15.15 (Homebrew) SET statement_timeout = 0; SET lock_timeout = 0; @@ -901,8 +901,8 @@ CREATE TABLE public.chat_messages ( sources jsonb[] DEFAULT ARRAY[]::jsonb[], suggestions text[] DEFAULT ARRAY[]::text[], feedback_type character varying(255), - CONSTRAINT valid_feedback_type CHECK ((((feedback_type)::text = ANY (ARRAY[('thumbs_up'::character varying)::text, ('thumbs_down'::character varying)::text])) OR (feedback_type IS NULL))), - CONSTRAINT valid_role CHECK (((role)::text = ANY (ARRAY[('user'::character varying)::text, ('assistant'::character varying)::text]))) + CONSTRAINT valid_feedback_type CHECK ((((feedback_type)::text = ANY ((ARRAY['thumbs_up'::character varying, 'thumbs_down'::character varying])::text[])) OR (feedback_type IS NULL))), + CONSTRAINT valid_role CHECK (((role)::text = ANY ((ARRAY['user'::character varying, 'assistant'::character varying])::text[]))) ); @@ -990,6 +990,99 @@ CREATE SEQUENCE public.clickhouse_query_executions_id_seq ALTER SEQUENCE public.clickhouse_query_executions_id_seq OWNED BY public.clickhouse_query_executions.id; +-- +-- Name: coinmarketcap_pro_backfill_assets; Type: TABLE; Schema: public; Owner: - +-- + +CREATE TABLE public.coinmarketcap_pro_backfill_assets ( + id bigint NOT NULL, + run_id bigint NOT NULL, + project_id bigint NOT NULL, + slug character varying(255) NOT NULL, + cmc_integer_id integer NOT NULL, + rank integer, + status character varying(255) DEFAULT 'pending'::character varying NOT NULL, + missing_ranges jsonb DEFAULT '{}'::jsonb NOT NULL, + points_exported integer DEFAULT 0 NOT NULL, + api_credits_used double precision DEFAULT 0.0 NOT NULL, + api_calls_total integer DEFAULT 0 NOT NULL, + rate_limited_calls_total integer DEFAULT 0 NOT NULL, + usage_precision character varying(255) DEFAULT 'exact'::character varying NOT NULL, + last_error text, + started_at timestamp(0) without time zone, + finished_at timestamp(0) without time zone, + inserted_at timestamp without time zone NOT NULL, + updated_at timestamp without time zone NOT NULL +); + + +-- +-- Name: coinmarketcap_pro_backfill_assets_id_seq; Type: SEQUENCE; Schema: public; Owner: - +-- + +CREATE SEQUENCE public.coinmarketcap_pro_backfill_assets_id_seq + START WITH 1 + INCREMENT BY 1 + NO MINVALUE + NO MAXVALUE + CACHE 1; + + +-- +-- Name: coinmarketcap_pro_backfill_assets_id_seq; Type: SEQUENCE OWNED BY; Schema: public; Owner: - +-- + +ALTER SEQUENCE public.coinmarketcap_pro_backfill_assets_id_seq OWNED BY public.coinmarketcap_pro_backfill_assets.id; + + +-- +-- Name: coinmarketcap_pro_backfill_runs; Type: TABLE; Schema: public; Owner: - +-- + +CREATE TABLE public.coinmarketcap_pro_backfill_runs ( + id bigint NOT NULL, + source character varying(255) DEFAULT 'coinmarketcap'::character varying NOT NULL, + scope character varying(255) NOT NULL, + status character varying(255) DEFAULT 'pending'::character varying NOT NULL, + "interval" character varying(255) DEFAULT '5m'::character varying NOT NULL, + time_start timestamp(0) without time zone NOT NULL, + time_end timestamp(0) without time zone NOT NULL, + dry_run boolean DEFAULT false NOT NULL, + total_assets integer DEFAULT 0 NOT NULL, + done_assets integer DEFAULT 0 NOT NULL, + failed_assets integer DEFAULT 0 NOT NULL, + pending_assets integer DEFAULT 0 NOT NULL, + api_credits_used_total double precision DEFAULT 0.0 NOT NULL, + api_calls_total integer DEFAULT 0 NOT NULL, + rate_limited_calls_total integer DEFAULT 0 NOT NULL, + usage_precision character varying(255) DEFAULT 'exact'::character varying NOT NULL, + last_error text, + started_at timestamp(0) without time zone, + finished_at timestamp(0) without time zone, + inserted_at timestamp without time zone NOT NULL, + updated_at timestamp without time zone NOT NULL +); + + +-- +-- Name: coinmarketcap_pro_backfill_runs_id_seq; Type: SEQUENCE; Schema: public; Owner: - +-- + +CREATE SEQUENCE public.coinmarketcap_pro_backfill_runs_id_seq + START WITH 1 + INCREMENT BY 1 + NO MINVALUE + NO MAXVALUE + CACHE 1; + + +-- +-- Name: coinmarketcap_pro_backfill_runs_id_seq; Type: SEQUENCE OWNED BY; Schema: public; Owner: - +-- + +ALTER SEQUENCE public.coinmarketcap_pro_backfill_runs_id_seq OWNED BY public.coinmarketcap_pro_backfill_runs.id; + + -- -- Name: comment_notifications; Type: TABLE; Schema: public; Owner: - -- @@ -5860,6 +5953,20 @@ ALTER TABLE ONLY public.classified_tweets ALTER COLUMN id SET DEFAULT nextval('p ALTER TABLE ONLY public.clickhouse_query_executions ALTER COLUMN id SET DEFAULT nextval('public.clickhouse_query_executions_id_seq'::regclass); +-- +-- Name: coinmarketcap_pro_backfill_assets id; Type: DEFAULT; Schema: public; Owner: - +-- + +ALTER TABLE ONLY public.coinmarketcap_pro_backfill_assets ALTER COLUMN id SET DEFAULT nextval('public.coinmarketcap_pro_backfill_assets_id_seq'::regclass); + + +-- +-- Name: coinmarketcap_pro_backfill_runs id; Type: DEFAULT; Schema: public; Owner: - +-- + +ALTER TABLE ONLY public.coinmarketcap_pro_backfill_runs ALTER COLUMN id SET DEFAULT nextval('public.coinmarketcap_pro_backfill_runs_id_seq'::regclass); + + -- -- Name: comment_notifications id; Type: DEFAULT; Schema: public; Owner: - -- @@ -6881,6 +6988,22 @@ ALTER TABLE ONLY public.clickhouse_query_executions ADD CONSTRAINT clickhouse_query_executions_pkey PRIMARY KEY (id); +-- +-- Name: coinmarketcap_pro_backfill_assets coinmarketcap_pro_backfill_assets_pkey; Type: CONSTRAINT; Schema: public; Owner: - +-- + +ALTER TABLE ONLY public.coinmarketcap_pro_backfill_assets + ADD CONSTRAINT coinmarketcap_pro_backfill_assets_pkey PRIMARY KEY (id); + + +-- +-- Name: coinmarketcap_pro_backfill_runs coinmarketcap_pro_backfill_runs_pkey; Type: CONSTRAINT; Schema: public; Owner: - +-- + +ALTER TABLE ONLY public.coinmarketcap_pro_backfill_runs + ADD CONSTRAINT coinmarketcap_pro_backfill_runs_pkey PRIMARY KEY (id); + + -- -- Name: comment_notifications comment_notifications_pkey; Type: CONSTRAINT; Schema: public; Owner: - -- @@ -8274,6 +8397,48 @@ CREATE INDEX classified_tweets_review_required_index ON public.classified_tweets CREATE INDEX clickhouse_query_executions_query_id_index ON public.clickhouse_query_executions USING btree (query_id); +-- +-- Name: coinmarketcap_pro_backfill_assets_run_id_project_id_index; Type: INDEX; Schema: public; Owner: - +-- + +CREATE UNIQUE INDEX coinmarketcap_pro_backfill_assets_run_id_project_id_index ON public.coinmarketcap_pro_backfill_assets USING btree (run_id, project_id); + + +-- +-- Name: coinmarketcap_pro_backfill_assets_run_id_rank_index; Type: INDEX; Schema: public; Owner: - +-- + +CREATE INDEX coinmarketcap_pro_backfill_assets_run_id_rank_index ON public.coinmarketcap_pro_backfill_assets USING btree (run_id, rank); + + +-- +-- Name: coinmarketcap_pro_backfill_assets_run_id_status_index; Type: INDEX; Schema: public; Owner: - +-- + +CREATE INDEX coinmarketcap_pro_backfill_assets_run_id_status_index ON public.coinmarketcap_pro_backfill_assets USING btree (run_id, status); + + +-- +-- Name: coinmarketcap_pro_backfill_assets_slug_index; Type: INDEX; Schema: public; Owner: - +-- + +CREATE INDEX coinmarketcap_pro_backfill_assets_slug_index ON public.coinmarketcap_pro_backfill_assets USING btree (slug); + + +-- +-- Name: coinmarketcap_pro_backfill_runs_inserted_at_index; Type: INDEX; Schema: public; Owner: - +-- + +CREATE INDEX coinmarketcap_pro_backfill_runs_inserted_at_index ON public.coinmarketcap_pro_backfill_runs USING btree (inserted_at); + + +-- +-- Name: coinmarketcap_pro_backfill_runs_status_index; Type: INDEX; Schema: public; Owner: - +-- + +CREATE INDEX coinmarketcap_pro_backfill_runs_status_index ON public.coinmarketcap_pro_backfill_runs USING btree (status); + + -- -- Name: contract_addresses_project_id_address_index; Type: INDEX; Schema: public; Owner: - -- @@ -9757,6 +9922,22 @@ ALTER TABLE ONLY public.clickhouse_query_executions ADD CONSTRAINT clickhouse_query_executions_user_id_fkey FOREIGN KEY (user_id) REFERENCES public.users(id) ON DELETE CASCADE; +-- +-- Name: coinmarketcap_pro_backfill_assets coinmarketcap_pro_backfill_assets_project_id_fkey; Type: FK CONSTRAINT; Schema: public; Owner: - +-- + +ALTER TABLE ONLY public.coinmarketcap_pro_backfill_assets + ADD CONSTRAINT coinmarketcap_pro_backfill_assets_project_id_fkey FOREIGN KEY (project_id) REFERENCES public.project(id) ON DELETE CASCADE; + + +-- +-- Name: coinmarketcap_pro_backfill_assets coinmarketcap_pro_backfill_assets_run_id_fkey; Type: FK CONSTRAINT; Schema: public; Owner: - +-- + +ALTER TABLE ONLY public.coinmarketcap_pro_backfill_assets + ADD CONSTRAINT coinmarketcap_pro_backfill_assets_run_id_fkey FOREIGN KEY (run_id) REFERENCES public.coinmarketcap_pro_backfill_runs(id) ON DELETE CASCADE; + + -- -- Name: comments comments_parent_id_fkey; Type: FK CONSTRAINT; Schema: public; Owner: - -- @@ -11057,7 +11238,7 @@ ALTER TABLE ONLY public.webinar_registrations -- PostgreSQL database dump complete -- -\unrestrict lPrBCVGzfDAQrezJuUuvsvvIxIMIDCrQVbcCW2A0GafidncDVtkqtUHNwYPqA5k +\unrestrict rEumk3or3c8G96Hxf37AhoShjhR5C9TGx5myvsAuvbIk1ywnIRepGcrCCCkez28 INSERT INTO public."schema_migrations" (version) VALUES (20171008200815); INSERT INTO public."schema_migrations" (version) VALUES (20171008203355); @@ -11579,6 +11760,7 @@ INSERT INTO public."schema_migrations" (version) VALUES (20250926101756); INSERT INTO public."schema_migrations" (version) VALUES (20250926115345); INSERT INTO public."schema_migrations" (version) VALUES (20251013121803); INSERT INTO public."schema_migrations" (version) VALUES (20251014144144); +INSERT INTO public."schema_migrations" (version) VALUES (20251015073648); INSERT INTO public."schema_migrations" (version) VALUES (20251016133413); INSERT INTO public."schema_migrations" (version) VALUES (20251017100000); INSERT INTO public."schema_migrations" (version) VALUES (20251021133911); @@ -11587,15 +11769,17 @@ INSERT INTO public."schema_migrations" (version) VALUES (20251023083446); INSERT INTO public."schema_migrations" (version) VALUES (20251023114153); INSERT INTO public."schema_migrations" (version) VALUES (20251027142731); INSERT INTO public."schema_migrations" (version) VALUES (20251027154645); +INSERT INTO public."schema_migrations" (version) VALUES (20251113070559); INSERT INTO public."schema_migrations" (version) VALUES (20251202143216); INSERT INTO public."schema_migrations" (version) VALUES (20251202143217); INSERT INTO public."schema_migrations" (version) VALUES (20251215114741); INSERT INTO public."schema_migrations" (version) VALUES (20251216081737); INSERT INTO public."schema_migrations" (version) VALUES (20260106131955); INSERT INTO public."schema_migrations" (version) VALUES (20260106141954); -INSERT INTO public."schema_migrations" (version) VALUES (20260114142311); INSERT INTO public."schema_migrations" (version) VALUES (20260114173809); INSERT INTO public."schema_migrations" (version) VALUES (20260116093636); INSERT INTO public."schema_migrations" (version) VALUES (20260216103643); INSERT INTO public."schema_migrations" (version) VALUES (20260224120000); INSERT INTO public."schema_migrations" (version) VALUES (20260225120000); +INSERT INTO public."schema_migrations" (version) VALUES (20260306173000); +INSERT INTO public."schema_migrations" (version) VALUES (20260306190000);