feat: automated push notifications via pg_cron and FCM (#52)

.env.example

@@ -44,3 +44,16 @@ NEXT_PUBLIC_WEATHER_API_KEY=your_openweathermap_api_key_here
 # Run `supabase start` then `supabase db reset` to create this test user.
 NEXT_PUBLIC_DEV_EMAIL=dev@agronavis.local
 NEXT_PUBLIC_DEV_PASSWORD=change-this-to-your-local-dev-password
+
+# -----------------------------------------------------------------------------
+# Firebase Configuration (Push Notifications)
+# -----------------------------------------------------------------------------
+NEXT_PUBLIC_FIREBASE_API_KEY=""
+NEXT_PUBLIC_FIREBASE_AUTH_DOMAIN=""
+NEXT_PUBLIC_FIREBASE_PROJECT_ID=""
+NEXT_PUBLIC_FIREBASE_STORAGE_BUCKET=""
+NEXT_PUBLIC_FIREBASE_MESSAGING_SENDER_ID=""
+NEXT_PUBLIC_FIREBASE_APP_ID=""
+NEXT_PUBLIC_FIREBASE_VAPID_KEY=""
+FIREBASE_PROJECT_ID=""
+FIREBASE_SERVICE_ACCOUNT=""

.gitignore

@@ -100,3 +100,4 @@ frontend/public/swe-worker-*.js
 # macOS resource forks
 ._*
 .DS_Store
+backend/supabase/.temp/

backend/main.py

@@ -25,12 +25,12 @@
 import torch.nn as nn
 from torchvision import models, transforms
 from PIL import Image
-from typing import List, Optional, Any, Dict
+from typing import List, Optional, Any, Dict, Literal
 from fastapi import FastAPI, File, UploadFile, HTTPException, Depends, Query
 from fastapi.middleware.cors import CORSMiddleware
 from chatbot import router as chatbot_router
 from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 from supabase import create_client, Client
 from dotenv import load_dotenv
 
@@ -207,6 +207,10 @@ class ProfileCreate(BaseModel):
     years_of_experience: Optional[int] = None
     education_level: Optional[str] = None
 
+class DeviceTokenCreate(BaseModel):
+    fcm_token: str = Field(min_length=1)
+    device_type: Literal["web", "android", "ios"] = "web"
+
 class ResourceCreate(BaseModel):
     farm_id: str
     resource_type: str
@@ -442,7 +446,21 @@ async def get_profile(user=Depends(verify_token)):
 async def upsert_profile(body: ProfileCreate, user=Depends(verify_token)):
     payload = {**body.model_dump(exclude_none=True), "id": user.id}
     res = supabase.table("farmers").upsert(payload).execute()
-    return {"success": True, "data": (res.data[0] if res.data else None)}
+
+# ── Device Tokens ──────────────────────────────────────────────────────────
+@app.post("/api/device-tokens", status_code=201)
+async def register_device_token(body: DeviceTokenCreate, user=Depends(verify_token)):
+    try:
+        res = supabase.rpc("upsert_device_token", {
+            "p_farmer_id": user.id,
+            "p_fcm_token": body.fcm_token,
+            "p_device_type": body.device_type,
+        }).execute()
+    except Exception as e:
+        if "TOKEN_OWNED_BY_ANOTHER_USER" in str(e):
+            raise HTTPException(status_code=403, detail="This device token is already registered to another account.")
+        raise
+    return {"success": True, "data": res.data}
 
 
 # ── Farms ─────────────────────────────────────────────────────────────────────

backend/supabase/functions/send-daily-tasks/index.ts

@@ -0,0 +1,183 @@
+import { serve } from "https://deno.land/std@0.168.0/http/server.ts";
+import { createClient } from "https://esm.sh/@supabase/supabase-js@2";
+
+const SUPABASE_URL = Deno.env.get("SUPABASE_URL")!;
+const SUPABASE_SERVICE_ROLE_KEY = Deno.env.get("SUPABASE_SERVICE_ROLE_KEY")!;
+const FIREBASE_PROJECT_ID = Deno.env.get("FIREBASE_PROJECT_ID")!;
+const FIREBASE_SERVICE_ACCOUNT = Deno.env.get("FIREBASE_SERVICE_ACCOUNT")!;
+
+function pemToBinary(pem: string): ArrayBuffer {
+  const cleaned = pem
+    .replace(/-----BEGIN PRIVATE KEY-----/, "")
+    .replace(/-----END PRIVATE KEY-----/, "")
+    .replace(/\s/g, "");
+  const binaryString = atob(cleaned);
+  const bytes = new Uint8Array(binaryString.length);
+  for (let i = 0; i < binaryString.length; i++) {
+    bytes[i] = binaryString.charCodeAt(i);
+  }
+  return bytes.buffer;
+}
+
+async function getFirebaseAccessToken(): Promise<string> {
+  const serviceAccount = JSON.parse(FIREBASE_SERVICE_ACCOUNT);
+
+  const header = { alg: "RS256", typ: "JWT" };
+  const now = Math.floor(Date.now() / 1000);
+
+  const payload = {
+    iss: serviceAccount.client_email,
+    sub: serviceAccount.client_email,
+    aud: "https://oauth2.googleapis.com/token",
+    iat: now,
+    exp: now + 3600,
+    scope: "https://www.googleapis.com/auth/firebase.messaging",
+  };
+
+  const encode = (obj: object) =>
+    btoa(JSON.stringify(obj)).replace(/=/g, "").replace(/\+/g, "-").replace(/\//g, "_");
+
+  const signingInput = `${encode(header)}.${encode(payload)}`;
+
+  const privateKey = await crypto.subtle.importKey(
+    "pkcs8",
+    pemToBinary(serviceAccount.private_key),
+    { name: "RSASSA-PKCS1-v1_5", hash: "SHA-256" },
+    false,
+    ["sign"]
+  );
+
+  const signature = await crypto.subtle.sign(
+    "RSASSA-PKCS1-v1_5",
+    privateKey,
+    new TextEncoder().encode(signingInput)
+  );
+
+  const jwt = `${signingInput}.${btoa(String.fromCharCode(...new Uint8Array(signature)))
+    .replace(/=/g, "").replace(/\+/g, "-").replace(/\//g, "_")}`;
+
+  const tokenRes = await fetch("https://oauth2.googleapis.com/token", {
+    method: "POST",
+    headers: { "Content-Type": "application/x-www-form-urlencoded" },
+    body: `grant_type=urn:ietf:params:oauth:grant-type:jwt-bearer&assertion=${jwt}`,
+  });
+
+  const tokenData = await tokenRes.json();
+  return tokenData.access_token;
+}
+
+async function sendFCMNotification(
+  accessToken: string,
+  fcmToken: string,
+  title: string,
+  body: string
+): Promise<{ ok: boolean; shouldDeactivate: boolean }> {
+  const res = await fetch(
+    `https://fcm.googleapis.com/v1/projects/${FIREBASE_PROJECT_ID}/messages:send`,
+    {
+      method: "POST",
+      headers: {
+        "Content-Type": "application/json",
+        Authorization: `Bearer ${accessToken}`,
+      },
+      body: JSON.stringify({
+        message: {
+          token: fcmToken,
+          notification: { title, body },
+          data: { click_action: "FLUTTER_NOTIFICATION_CLICK" },
+        },
+      }),
+    }
+  );
+
+  if (res.ok) return { ok: true, shouldDeactivate: false };
+
+  const errorBody = await res.json().catch(() => null);
+  const errorStatus = errorBody?.error?.status;
+  console.error(`FCM send failed for token ${fcmToken.slice(0, 12)}...: ${res.status} ${errorStatus ?? ""}`);
+
+  const shouldDeactivate = errorStatus === "UNREGISTERED" || errorStatus === "INVALID_ARGUMENT" || errorStatus === "NOT_FOUND";
+  return { ok: false, shouldDeactivate };
+}
+
+serve(async () => {
+  const supabase = createClient(SUPABASE_URL, SUPABASE_SERVICE_ROLE_KEY);
+  const today = new Date().toISOString().split("T")[0];
+
+  const { data: tasks } = await supabase
+    .from("farm_tasks")
+    .select("id, title, description, farmer_id:farms(farmer_id)")
+    .eq("due_date", today)
+    .eq("status", "pending");
+
+  const { data: harvests } = await supabase
+    .from("crops")
+    .select("id, crop_type, variety, expected_harvest_date, farmer_id:farms(farmer_id)")
+    .eq("expected_harvest_date", today);
+
+  if ((!tasks || tasks.length === 0) && (!harvests || harvests.length === 0)) {
+    return new Response(JSON.stringify({ message: "No notifications to send today" }), {
+      status: 200,
+      headers: { "Content-Type": "application/json" },
+    });
+  }
+
+  const accessToken = await getFirebaseAccessToken();
+
+  const farmerNotifications: Record<string, { titles: string[]; bodies: string[] }> = {};
+
+  for (const task of tasks ?? []) {
+    const farmerId = task.farmer_id?.farmer_id;
+    if (!farmerId) continue;
+    if (!farmerNotifications[farmerId]) farmerNotifications[farmerId] = { titles: [], bodies: [] };
+    farmerNotifications[farmerId].titles.push(task.title);
+    farmerNotifications[farmerId].bodies.push(task.description ?? task.title);
+  }
+
+  for (const crop of harvests ?? []) {
+    const farmerId = crop.farmer_id?.farmer_id;
+    if (!farmerId) continue;
+    if (!farmerNotifications[farmerId]) farmerNotifications[farmerId] = { titles: [], bodies: [] };
+    farmerNotifications[farmerId].titles.push("Harvest Ready");
+    farmerNotifications[farmerId].bodies.push(`Your ${crop.crop_type} (${crop.variety}) is ready for harvest today.`);
+  }
+
+  let sent = 0;
+  let failed = 0;
+  const tokensToDeactivate: string[] = [];
+
+  for (const [farmerId, notif] of Object.entries(farmerNotifications)) {
+    const { data: tokens } = await supabase
+      .from("device_tokens")
+      .select("fcm_token")
+      .eq("farmer_id", farmerId)
+      .eq("is_active", true);
+
+    if (!tokens || tokens.length === 0) continue;
+
+    const title = notif.titles.length === 1 ? notif.titles[0] : `You have ${notif.titles.length} tasks today`;
+    const body = notif.bodies.slice(0, 2).join(" | ");
+
+    for (const { fcm_token } of tokens) {
+      const result = await sendFCMNotification(accessToken, fcm_token, title, body);
+      if (result.ok) {
+        sent++;
+      } else {
+        failed++;
+        if (result.shouldDeactivate) tokensToDeactivate.push(fcm_token);
+      }
+    }
+  }
+
+  if (tokensToDeactivate.length > 0) {
+    await supabase
+      .from("device_tokens")
+      .update({ is_active: false })
+      .in("fcm_token", tokensToDeactivate);
+  }
+
+  return new Response(JSON.stringify({ message: "Notifications processed", sent, failed }), {
+    status: 200,
+    headers: { "Content-Type": "application/json" },
+  });
+});

backend/supabase/migrations/20260619000000_push_notifications_pg_cron.sql

@@ -0,0 +1,59 @@
+CREATE EXTENSION IF NOT EXISTS pg_cron WITH SCHEMA extensions;
+CREATE EXTENSION IF NOT EXISTS pg_net  WITH SCHEMA extensions;
+
+CREATE TABLE IF NOT EXISTS public.device_tokens (
+    id          uuid        NOT NULL DEFAULT gen_random_uuid(),
+    farmer_id   uuid        NOT NULL REFERENCES public.farmers(id) ON DELETE CASCADE,
+    fcm_token   text        NOT NULL,
+    device_type text        NOT NULL DEFAULT 'web'
+                            CHECK (device_type = ANY (ARRAY['web', 'android', 'ios'])),
+    is_active   boolean     NOT NULL DEFAULT true,
+    created_at  timestamptz NOT NULL DEFAULT now(),
+    updated_at  timestamptz NOT NULL DEFAULT now(),
+    CONSTRAINT device_tokens_pkey         PRIMARY KEY (id),
+    CONSTRAINT device_tokens_token_unique UNIQUE (fcm_token)
+);
+
+CREATE INDEX IF NOT EXISTS idx_device_tokens_farmer_id
+    ON public.device_tokens (farmer_id)
+    WHERE is_active = true;
+
+CREATE TRIGGER set_device_tokens_updated_at
+    BEFORE UPDATE ON public.device_tokens
+    FOR EACH ROW
+    EXECUTE FUNCTION update_updated_at_column();
+
+ALTER TABLE public.device_tokens ENABLE ROW LEVEL SECURITY;
+
+CREATE POLICY "Farmers can view own device tokens"
+    ON public.device_tokens FOR SELECT
+    USING (auth.uid() = farmer_id);
+
+CREATE POLICY "Farmers can insert own device tokens"
+    ON public.device_tokens FOR INSERT
+    WITH CHECK (auth.uid() = farmer_id);
+
+CREATE POLICY "Farmers can update own device tokens"
+    ON public.device_tokens FOR UPDATE
+    USING (auth.uid() = farmer_id);
+
+CREATE POLICY "Farmers can delete own device tokens"
+    ON public.device_tokens FOR DELETE
+    USING (auth.uid() = farmer_id);
+
+GRANT SELECT, INSERT, UPDATE, DELETE ON public.device_tokens TO service_role;
+
+SELECT cron.schedule(
+    'send-daily-task-notifications',
+    '30 0 * * *',
+    $$
+    SELECT net.http_post(
+        url     := 'https://<YOUR_PROJECT_REF>.supabase.co/functions/v1/send-daily-tasks',
+        headers := jsonb_build_object(
+            'Content-Type',  'application/json',
+            'Authorization', 'Bearer ' || current_setting('app.service_role_key', true)
+        ),
+        body    := '{}'::jsonb
+    );
+    $$
+);

backend/supabase/migrations/20260620000000_device_token_upsert_function.sql

@@ -0,0 +1,31 @@
+CREATE OR REPLACE FUNCTION public.upsert_device_token(
+    p_farmer_id uuid,
+    p_fcm_token text,
+    p_device_type text
+)
+RETURNS public.device_tokens
+LANGUAGE plpgsql
+SECURITY DEFINER
+SET search_path = public
+AS $$
+DECLARE
+    v_row public.device_tokens;
+BEGIN
+    INSERT INTO public.device_tokens (farmer_id, fcm_token, device_type, is_active)
+    VALUES (p_farmer_id, p_fcm_token, p_device_type, true)
+    ON CONFLICT (fcm_token) DO UPDATE
+        SET device_type = EXCLUDED.device_type,
+            is_active = true,
+            updated_at = now()
+        WHERE public.device_tokens.farmer_id = p_farmer_id
+    RETURNING * INTO v_row;
+
+    IF v_row.id IS NULL THEN
+        RAISE EXCEPTION 'TOKEN_OWNED_BY_ANOTHER_USER' USING ERRCODE = '23505';
+    END IF;
+
+    RETURN v_row;
+END;
+$$;
+
+GRANT EXECUTE ON FUNCTION public.upsert_device_token(uuid, text, text) TO service_role;