RWM Console cluster: risingwave-alinma.alinma-rw.svc.cluster.local

CREATE MATERIALIZED VIEW alinma_bff.unlinked_account_pairs_mv AS
WITH pair_candidates AS (
  SELECT
    sec.account_id AS security_account_id,
    cash.account_id AS cash_account_id,
    ri.id AS reference_identifier_id,
    JSONB_BUILD_OBJECT(
      'id',
      sec.account_id,
      'external_id',
      '',
      'account_number',
      sec.number,
      'currency_code',
      COALESCE(sec.base_currency_code, ''),
      'opened_at',
      CASE
        WHEN sec.opening_date IS NULL
        THEN NULL
        ELSE JSONB_BUILD_OBJECT('seconds', CAST(EXTRACT(EPOCH FROM CAST(sec.opening_date AS TIMESTAMP)) AS BIGINT))
      END,
      'client',
      sec_client.client_json
    ) AS security_account,
    JSONB_BUILD_OBJECT(
      'id',
      cash.account_id,
      'external_id',
      '',
      'account_number',
      cash.number,
      'currency_code',
      COALESCE(cash.base_currency_code, ''),
      'opened_at',
      CASE
        WHEN cash.opening_date IS NULL
        THEN NULL
        ELSE JSONB_BUILD_OBJECT(
          'seconds',
          CAST(EXTRACT(EPOCH FROM CAST(cash.opening_date AS TIMESTAMP)) AS BIGINT)
        )
      END,
      'client',
      cash_client.client_json
    ) AS cash_account
  FROM olap.reference_identifiers AS ri
  JOIN olap.accounts_dm AS sec
    ON sec.account_id = ri.entity_id AND sec.disabled_at IS NULL
  JOIN olap.product_types_dm AS sec_pt
    ON sec_pt.product_type_id = sec.product_type_id AND sec_pt.type = 'INVESTMENT'
  JOIN olap.accounts_dm AS cash
    ON cash.number = ri.value AND cash.disabled_at IS NULL
  JOIN alinma_bff.account_primary_client_mv AS sec_client
    ON sec_client.account_id = sec.account_id
  JOIN alinma_bff.account_primary_client_mv AS cash_client
    ON cash_client.account_id = cash.account_id
  LEFT JOIN olap.account_to_portfolios_dm AS ap
    ON ap.account_id = sec.account_id
    AND ap.disabled_at IS NULL
    AND CAST(ap.effective_start_date AS TIMESTAMPTZ) <= CURRENT_TIMESTAMP
    AND CAST(COALESCE(ap.effective_end_date, CAST('9999-12-31' AS DATE)) AS TIMESTAMPTZ) > CURRENT_TIMESTAMP
  WHERE
    ri.entity_type = 'account'
    AND ri.key = 'cash_account_num'
    AND ap.account_id IS NULL
)
SELECT
  ID_FROM_STRING_WITH_PREFIX('unlinked_account_pair', security_account_id || ':' || cash_account_id) AS id,
  security_account_id,
  cash_account_id,
  'risingwave-pipeline' AS registered_by,
  (
    JSONB_AGG(security_account ORDER BY reference_identifier_id) -> 0
  ) AS security_account,
  (
    JSONB_AGG(cash_account ORDER BY reference_identifier_id) -> 0
  ) AS cash_account
FROM pair_candidates
GROUP BY
  security_account_id,
  cash_account_id