前提と注意事項

この記事では以下を前提としています。

• Supabase の pg_cron、pg_net、Vault を使用しています。
• Next.js 16.4.2 を Vercel 上で動かしています。
• ジョブを管理する表 document_jobs と、ジョブを取り出す関数 claim_document_jobs() が作成済みです。この記事では、これらの定義は扱いません。
• 記事中のドメインは例として app.example.com に置き換えています。ご自身の環境のドメインに読み替えてください。

この記事のゴール

pg_cron から毎分呼ぶ関数が、拾えるジョブがあるときだけ pg_net で Next.js の内部 API を呼び出し、認証付きで処理が動くところまでをゴールとします。

ここで扱う document_jobs は、ジョブの表です。たまった処理を Next.js の内部 API（POST /api/internal/document-jobs）で実行します。

Webhook を受けた直後は、Next.js の after() で同じ処理を動かします。そのため、cron は取りこぼしを拾うための保険という位置づけです。1分以内の遅れは許容します。

ただし、毎分 HTTP リクエストを送ると、ほとんどの回は空振りになります。そこで、拾えるジョブがあるときだけ POST するようにします。

構成概要

全体の流れは以下のとおりです。

pg_cron（毎分）
  ↓ select invoke_document_jobs()
invoke_document_jobs()：拾えるジョブがなければ何もしない
  ↓ pg_net（net.http_post）、Authorization: Bearer <シークレット>
Next.js の内部 API（/api/internal/document-jobs）
  ↓ claim_document_jobs() で取り出して処理

呼び出す関数を作成する

まず、拾えるジョブの有無を確認し、あるときだけ内部 API に POST する関数 invoke_document_jobs() を作成します。URL とシークレットは Vault から取得します。

create or replace function invoke_document_jobs()
returns void
language plpgsql
security definer
set search_path = public, extensions, vault
as $$
declare
  v_url        text;
  v_secret     text;
  v_request_id bigint;
begin
  -- 拾えるジョブ（queued で run_after が来た、または running でリースが切れた）がなければ呼ばない
  if not exists (
    select 1 from document_jobs j
     where j.kind = 'fetch_email'
       and ((j.status = 'queued' and j.run_after <= now() and j.attempts < j.max_attempts)
            or (j.status = 'running' and j.lease_until < now()))
  ) then
    return;
  end if;
 
  select decrypted_secret into v_url
    from vault.decrypted_secrets where name = 'document_jobs_url' limit 1;
  select decrypted_secret into v_secret
    from vault.decrypted_secrets where name = 'internal_function_secret' limit 1;
 
  if v_url is null or v_secret is null then
    raise warning '[document-jobs] vault secrets missing; skipping';
    return;
  end if;
 
  select net.http_post(
    url                  := v_url,
    body                 := '{}'::jsonb,
    headers              := jsonb_build_object(
                              'Content-Type',  'application/json',
                              'Authorization', 'Bearer ' || v_secret
                            ),
    timeout_milliseconds := 300000
  ) into v_request_id;
end;
$$;
 
revoke all on function invoke_document_jobs() from public;
revoke all on function invoke_document_jobs() from anon, authenticated;

関数の最後の revoke で、一般のロールから実行できないように権限を絞っています。

HTTP POST を送る方法については、以下の記事でも解説しています。

拾えるジョブがあるときだけ呼ぶ条件

exists の条件では、次の2種類のジョブを「拾えるジョブ」として扱います。

• queued で、run_after の時刻が来ており、試行回数が上限に達していないジョブ
• running のまま、リース（lease_until）が切れたジョブ

リースが切れたジョブは、試行回数が上限に達していても条件に含めています。claim_document_jobs() がそのジョブを failed に片付けるためです。その掃除のためだけにでも、API を呼ぶ必要があります。

Vault の値が未登録のときの動作

URL かシークレットのどちらかが Vault から取得できない場合は、warning を出して POST せずに終了します。そのため、Vault に登録する前でもエラーにはなりません。

Vault に URL とシークレットを登録する

内部 API の URL とシークレットを、Supabase の SQL エディタから Vault に登録します。値はリポジトリに入れません。

select vault.create_secret('https://app.example.com/api/internal/document-jobs', 'document_jobs_url');
select vault.create_secret('<ランダムな長い文字列>', 'internal_function_secret');

https://app.example.com/... の部分はご自身の環境の URL に、<ランダムな長い文字列> はご自身で生成した値に置き換えてください。この値は、後述する Next.js の環境変数 INTERNAL_FUNCTION_SECRET と同じにします。

Vault に登録するまでは、関数は warning を出すだけで POST しません。そのため、先にマイグレーションを当てておき、デプロイの後に URL を登録する順にできます。

Vault の使い方については、以下の記事で解説しています。

pg_cron にジョブを登録する

毎分 invoke_document_jobs() を実行するように、pg_cron にジョブを登録します。マイグレーションを何度当てても重複しないように、同じ名前のジョブを消してから登録します。

do $$
declare
  j record;
begin
  for j in select jobid from cron.job where jobname = 'document-jobs' loop
    perform cron.unschedule(j.jobid);
  end loop;
 
  perform cron.schedule(
    'document-jobs',
    '* * * * *',
    $cron$ select public.invoke_document_jobs(); $cron$
  );
end $$;

pg_cron の基本的な使い方は、以下の記事で解説しています。

Next.js の内部 API で認証する

内部 API では、Authorization: Bearer <シークレット> の値を、環境変数 INTERNAL_FUNCTION_SECRET と定数時間で比較します。比較には timingSafeEqual を使います。一致しない場合は 401 を返します。

まず、認証を判定する関数です。

import { timingSafeEqual } from "node:crypto";
import type { NextRequest } from "next/server";
 
export function isInternalRequestAuthorized(req: NextRequest, logPrefix: string): boolean {
  const secret = process.env.INTERNAL_FUNCTION_SECRET;
  if (!secret) {
    // 401 だけだと Vault の値違いと区別がつかない。cron は「成功」と記録するのでここで残す
    console.error(`${logPrefix} INTERNAL_FUNCTION_SECRET is not set; rejecting cron call`);
    return false;
  }
  const bearer = bearerToken(req); // Authorization: Bearer <値> の値。なければ null
  if (!bearer) return false;
  const given = Buffer.from(bearer);
  const expected = Buffer.from(secret);
  return given.length === expected.length && timingSafeEqual(given, expected);
}

bearerToken() は、Authorization ヘッダーの Bearer に続く値を取り出し、なければ null を返す関数です。

次に、route で認証に通らなければ 401 を返します。

export const maxDuration = 300;
 
export async function POST(req: NextRequest) {
  if (!isInternalRequestAuthorized(req, "[document-jobs]")) {
    return NextResponse.json({ error: "unauthorized" }, { status: 401 });
  }
  // claim_document_jobs() で取り出して処理する
}

タイムアウトを合わせる

pg_net の timeout_milliseconds は、route の maxDuration（300秒）に合わせて 300000 にしています。pg_net 側のタイムアウトが短いと、Next.js 側の処理が終わる前に待つのをやめてしまうためです。

トラブルシューティング

401 が返るが原因が分からない

INTERNAL_FUNCTION_SECRET が未設定のときは、401 を返すだけでなく、エラーログを残すようにしています。401 だけでは、Vault の値の間違いと区別がつかないためです。

また、cron 側は POST を積んだ時点で「成功」と記録します。そのため、cron の実行履歴からは認証の失敗に気づけません。Next.js 側のログを確認してください。

呼んだのに何も拾わない、または拾えるのに呼ばれない

空振りを避けるための exists の条件は、claim_document_jobs() の取り出し条件と一致させる必要があります。両者がずれると、「呼んだのに何も拾わない」か「拾えるのに呼ばれない」状態になります。どちらかの条件を変更したときは、もう一方も見直してください。

pg_net の結果が cron 側で分からない

pg_net は非同期で動作します。net.http_post は要求を積んで ID を返すだけです。応答は net._http_response に入ります。cron の関数の中では、結果を待ちません。

cron と after() が同時に動く

cron と、Webhook の後の after() が同時に動く場合があります。その場合も、claim_document_jobs() の for update skip locked により、同じジョブを取らないようにしています。

まとめ

この記事では、pg_cron から毎分呼ぶ関数の中で、拾えるジョブがあるときだけ pg_net で Next.js の内部 API を呼ぶ仕組みを解説しました。ポイントを整理すると以下のとおりです。

• exists で拾えるジョブの有無を確認し、空振りの HTTP リクエストを避けます。
• exists の条件は、claim_document_jobs() の取り出し条件と一致させます。
• URL とシークレットは Vault に登録し、リポジトリには入れません。
• 内部 API では、timingSafeEqual による定数時間の比較で認証します。
• pg_net の timeout_milliseconds は、route の maxDuration に合わせます。

定期的に実行したい処理を、必要なときだけ Next.js の API で動かしたい方にとって参考になれば幸いです。