ซิงโครไนซ์ Redis State กับ Database ได้อย่างราบรื่นด้วย QStash

ในบทความก่อนหน้า เราได้พูดถึงการใช้ Redis เก็บ state ของ user quota บน full-stack Next.js app การทำอย่างนี้ช่วยเพิ่มประสิทธิภาพแอปพลิเคชัน เนื่องจาก Redis โดยทั่วไปเร็วกว่าฐานข้อมูลมาก

Upstash Redis เหมาะสำหรับ use case นี้เป็นพิเศษ เนื่องจากเป็น fully managed Redis database และเข้าถึงได้ผ่าน HTTP หมายความว่าเราสามารถใช้ edge runtimes เช่น Vercel รันโค้ดใกล้ผู้ใช้ด้วย latency ต่ำสุด

QStash คืออะไร?

QStash เป็น HTTP-based messaging และ scheduling solution สำหรับ serverless และ edge runtimes กล่าวคือ มันช่วยให้คุณ รัน CRON jobs โดยส่ง HTTP requests

CRON jobs เป็นวิธียอดเยี่ยมในการ schedule tasks ให้รันในเวลาที่กำหนด CRON job นี้ใช้เวลานานแค่ไหนก็ได้โดยไม่กระทบประสิทธิภาพแอปพลิเคชันหรือ end-users

เป้าหมายของ CRON job ในบทความนี้คือ sync state ของ user quota จาก Redis ไปยัง database เราจะทำให้ task นี้ รันทุกชั่วโมง

สร้าง CRON job endpoint ด้วย QStash

สรุปย้อนหลังจากบทความก่อน วิธีจัดการ state ของ user quota ใน Redis:

// Key stored in Redis. สร้าง key สำหรับแต่ละ user unique ด้วย ID
const quotaKey = `user:${userId}:quota`;

// เรียกคำสั่ง INCR เพื่อเพิ่มค่า key ทีละ 1
const quota = await redis.incr(quotaKey);

// ถ้าค่า key เป็น 1 หมายถึง key ถูกสร้างใหม่
// ตั้ง expiration date 24 ชั่วโมงด้วยคำสั่ง EXPIRE
if (quota === 1) {
  await redis.expire(quotaKey, 60 * 60 * 24);
}

จากโค้ดนี้ เราสามารถสร้าง CRON job endpoint ด้วย QStash ได้ CRON job endpoint จะถูก trigger ที่ URL /api/cron/update-usage

สร้างไฟล์ใหม่สำหรับ Next.js endpoint:

touch pages/api/cron/update-usage.ts

จากนั้น ต้องแน่ใจว่าแค่ QStash เท่านั้นที่ trigger endpoint นี้ได้ นี่คือสิ่งที่ @upstash/qstash จัดการ ถ้าไม่ทำเช่นนี้ ใครก็อาจ potentially trigger CRON job endpoint นำไปสู่ security issues

ติดตั้ง @upstash/qstash:

npm install @upstash/qstash

ตอนนี้ focus ที่โค้ด CRON job endpoint

สรุปย้อนหลังสิ่งที่เราจะทำ:

• ดึง users' quota keys ทั้งหมดจาก Redis ด้วย SCAN (performant cursor-based pagination)
• สำหรับแต่ละ key จาก SCAN สร้าง array users ของ objects ที่มี user ID และ quota
• สร้าง transaction สำหรับแต่ละ user ใน array users อัปเดต quota ใน database ใช้ Prisma
• Export endpoint function ด้วย verifySignature จาก @upstash/qstash/nextjs เพื่อรับรองว่าแค่ QStash เท่านั้นที่ trigger endpoint

โค้ด endpoint เต็มอยู่ที่นี่ ผม comment ทุกส่วนให้ชัดเจน:

// /pages/api/cron/update-usage.ts
import type { NextApiRequest, NextApiResponse } from "next";
import { verifySignature } from "@upstash/qstash/nextjs";
import prisma from "@/lib/prisma";

// Generate Upstash Redis instance จาก environment variables
// ตรวจสอบว่า defined ใน .env file แล้ว
const redis = new Redis({
  url: process.env.UPSTASH_REDIS_REST_URL!,
  token: process.env.UPSTASH_REDIS_REST_TOKEN!,
});

// Required by `@upstash/qstash`
export const config = {
  api: {
    bodyParser: false
  }
};

// อัปเดต usage ของทุก user ดึงข้อมูลจาก Redis database เข้า Prisma database
async function handler(req: NextApiRequest, res: NextApiResponse) {
  // Contains all the keys found during the `SCAN`
  const keys: string[] = [];
  
  // Current position of the cursor, updated after each `SCAN`
  let cursor = 0;
  
  // Execute cursor-based pagination ใช้คำสั่ง `SCAN`
  // Redis return `0` เมื่อ cursor อยู่ที่สิ้นสุด pagination
  do {
    const [nextCursor, newKeys] = await redis.scan(cursor, {
      // Match pattern ของ keys ที่ต้องการหา * คือ wildcard
      match: "user:*:quota",
      // Limit จำนวน keys ที่ SCAN คืนค่า
      count: 10,
    });
    
    cursor = nextCursor;
    keys.push(...newKeys);
  } while (cursor !== 0);
  
  // Array of objects containing the user's ID and their quota
  const users: { id: number; quota: number }[] = [];
  
  // ดึง usage ของแต่ละ user จาก Redis ตาม keys ที่หาได้
  for (const key of keys) {
    // Key ควรอยู่ใน format `user:${id}:quota` split key เพื่อได้ ID
    const id = key.split(":")[1];
    
    // ดึง quota ปัจจุบันของ user ด้วยคำสั่ง `GET`
    const quota = await redis.get(`user:${id}:quota`);
    
    users.push({ id: parseInt(id, 10), quota });
  }
  
  // สร้าง transaction สำหรับแต่ละ user อัปเดต quota ใน database
  const promises = users.map((user) => 
    prisma.user.update({
      where: { id: user.id },
      data: { quota: user.quota },
    }),
  );
  
  // รัน transactions ทั้งหมด
  await Promise.all(promises);
  
  return res.status(200).json({ message: "OK" });
}

// Export handler ด้วย `verifySignature` function เพื่อให้แน่ใจว่าแค่ QStash เท่านั้นที่ authenticate trigger CRON job
export default verifySignature(handler);

ส่วนยากที่สุดเสร็จแล้ว!

ตั้งค่า QStash

ก่อน push โค้ด ต้อง setup QStash QStash มี free tier ให้ใช้ได้ถึง 500 requests ต่อวัน

Login เข้า Upstash Console คลิกแท็บ QStash

จากนั้น สร้าง scheduled CRON job ใหม่ใช้ Request Builder ง่ายมากด้วย UI ที่ intuitive:

หลังคลิก Schedule เลื่อนลง จะเห็น CRON job ในส่วน Scheduled Jobs:

ขั้นสุดท้าย ต้อง setup 2 environment variables ใน Next.js application ตัวแปรเหล่านี้จะเพิ่มที่ host Next.js app ในที่นี้คือ Vercel

ในส่วน Request Builder มี 2 environment variables พร้อม copy:

• QSTASH_CURRENT_SIGNING_KEY
• QSTASH_NEXT_SIGNING_KEY

ตัวแปรเหล่านี้จำเป็นสำหรับ authenticate incoming message ไม่เช่นนั้นใครก็จะเรียก endpoint ของคุณได้

ได้ตัวแปรแล้ว เพิ่มลงใน Vercel project ไปที่ Vercel project คลิก Settings แล้ว Environment Variables:

หลังเพิ่ม 2 environment variables แล้ว สามารถ push โค้ดได้

QStash จะ trigger CRON job endpoint เป็นระยะ อัปเดต usage ของทุก user ทุกชั่วโมง

สรุป

Upstash มีเครื่องมือมากมายสำหรับ serverless environment เราสามารถใช้ QStash และ Redis ร่วมกันสร้างแอปพลิเคชันที่ performant และ scalable สำหรับ use cases หลายอย่าง

อยากดู live implementation ระบบเดียวกัน implement บน fastsheet tool ที่แปลง spreadsheets เป็น API