একটি লোডিং স্পিনার আপনাকে কিছুই জানাতে পারে না। যখন একটি AI টাস্ক সম্পন্ন হতে কয়েক মিনিট সময় নেয়—অথবা তৃতীয়বার চেষ্টা করার জন্য কিউতে (queue) ফিরে যায়—তখন আপনার স্ট্যাটাস বা অবস্থা দেখা প্রয়োজন। Server-Sent Events আপনাকে WebSockets-এর হ্যান্ডশেক ওভারহেড বা long polling-এর জটিলতা ছাড়াই সেই ভিজিবিলিটি প্রদান করে। সার্ভার একটি মাত্র HTTP রেসপন্স খোলা রাখে এবং পরিবর্তন অনুযায়ী প্লেইন-টেক্সট আপডেট পাঠাতে থাকে। ক্লায়েন্ট সেগুলো আসার সাথে সাথেই পড়ে নেয়।

যদি কানেকশন বিচ্ছিন্ন হয়ে যায়, তবে আপনি সম্ভবত সবকিছু আবার শুরু করতে চাইবেন না। একটি সুগঠিত SSE স্ট্রিম মনে রাখে আপনি কোথায় ছিলেন। শুধুমাত্র Node.js 20 এবং এর স্ট্যান্ডার্ড লাইব্রেরি ব্যবহার করেই আপনি এটি তৈরি করতে পারেন। কোনো এক্সটার্নাল প্যাকেজের প্রয়োজন নেই।

ওয়্যার ফরম্যাটটি দেখতে কেমন

একটি SSE মেসেজ হলো সাধারণ টেক্সট। সার্ভার তিনটি জিনিস লেখে: একটি ঐচ্ছিক ইভেন্ট নাম (event name), একটি প্রয়োজনীয় data ফিল্ড, এবং একটি id ফিল্ড যা আপনার সেভ পয়েন্ট হিসেবে কাজ করে। প্রতিটি রেকর্ড দুটি নিউলাইন ক্যারেক্টার দিয়ে শেষ হয়—একটি ফাঁকা লাইন যা সীমানা চিহ্নিত করে।

একটি সুস্থ স্ট্রিম ওয়্যারে দেখতে অনেকটা এরকম হতে পারে:

id: 14
event: status
data: {"phase":"testing","progress":43}

id: 15
event: status
data: {"phase":"retrying","attempt":2}

ব্রাউজারের EventSource ক্লায়েন্ট স্বয়ংক্রিয়ভাবে এই লাইনগুলো পড়ে নেয়। এটি প্রতিটি ব্লকের জন্য একটি ইভেন্ট তৈরি করে এবং অভ্যন্তরীণভাবে সর্বশেষ id সংরক্ষণ করে। যদি TCP কানেকশন বিচ্ছিন্ন হয়ে যায়, ক্লায়েন্ট অপেক্ষা করে, পুনরায় কানেক্ট করে এবং সংরক্ষিত আইডেন্টিফায়ারটি Last-Event-ID হেডার হিসেবে সার্ভারে ফেরত পাঠায়। এই হেডারটিই হলো এই প্যাটার্নটি কাজ করার মূল কারণ। এটি ছাড়া আপনার কাছে কোনো স্থায়ী কার্সার (durable cursor) থাকবে না।

Node.js-এ সার্ভার কানেক্ট করা

Node-এর বিল্ট-ইন http মডিউল সরাসরি এটি হ্যান্ডেল করতে পারে। যখন একটি রিকোয়েস্ট আসে, তখন সঠিক হেডার সেট করুন যাতে ক্লায়েন্ট বুঝতে পারে এটি একটি স্ট্রিম, কোনো পেজ নয়:

Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive

বাফারিং (buffering) সরিয়ে ফেলুন। প্রক্সি এবং ফ্রেমওয়ার্কগুলো মাঝে মাঝে রেসপন্স ব্যাচ আকারে পাঠায়, যা রিয়েল-টাইম অনুভূতি নষ্ট করে দেয়, তাই প্রতিটি চাঙ্ক (chunk) পাঠানোর পর তা ফ্ল্যাশ (flush) করে দিন।

প্রথমে ID পাঠান, তারপর ইভেন্ট টাইপ, তারপর পেলোড ডেটা এবং সবশেষে টার্মিনেটিং ফাঁকা লাইনটি পাঠান। এখানে ক্রম বা অর্ডার গুরুত্বপূর্ণ কারণ ফাঁকা লাইনের আগে অবশ্যই ID পৌঁছাতে হবে যাতে ক্লায়েন্ট সেটি ক্যাপচার করতে পারে। আপনি যদি নেটিভ response.write() ব্যবহার করেন, তবে আউটপুটটি হবে ঠিক এরকম:

response.write(`id: ${cursor}\n`);
response.write(`event: ${eventName}\n`);
response.write(`data: ${JSON.stringify(payload)}\n\n`);

শেষের ওই \n\n কেবল সাজানোর জন্য নয়। SSE পার্সারগুলো এটিকে রেকর্ড টার্মিনেটর হিসেবে গণ্য করে। এটি বাদ পড়লে ক্লায়েন্ট আরও ডেটার জন্য অপেক্ষা করতে করতে আটকে (hang) থাকবে।

কার্সারই হলো সবকিছু

একটি নতুন HTTP কানেকশন নতুন স্টেট বা অবস্থার নিশ্চয়তা দেয় না। যখন একটি ক্লায়েন্ট পুনরায় কানেক্ট করে, তখন Last-Event-ID হেডার আপনাকে বলে দেয় তারা সর্বশেষ কোন মেসেজটি পেয়েছিল। আপনার কাজ হলো পরবর্তী মেসেজ থেকে শুরু করা, শুরু থেকে নয়।

এর মানে হলো সার্ভার-সাইডে ইভেন্টগুলোর একটি ক্রমানুসারী লগ বা জার্নাল বজায় রাখা। একটি ডেমোর জন্য ইন-মেমরি অ্যারে (in-memory array) কাজ করতে পারে। কিন্তু প্রোডাকশনে আপনার এমন কিছু প্রয়োজন যা স্থায়ী—যেমন একটি ডাটাবেস লগ, Redis স্ট্রিম বা write-ahead জার্নালে ডেটা যুক্ত করা—কারণ সার্ভার রিস্টার্ট হলে যেন ইতিহাস মুছে না যায় এবং প্রতিটি ক্লায়েন্টকে আবার শূন্য থেকে শুরু করতে না হয়।

আপনার ইভেন্টগুলোকে একটি monotonically increasing integer বা ULID দিয়ে ইনডেক্স করুন। যখন রিকানেক্ট রিকোয়েস্ট আসে, তখন id > lastEventId এমন ইভেন্টগুলো কুয়েরি করুন এবং সেগুলো ক্রমানুসারে প্লে (replay) করুন। যদি আপনার কাছে শত শত ব্যাকলগ মেসেজ থাকে, তবে একটি ছোট কৃত্রিম বিলম্ব (delay) বা ব্যাচিং ব্যবহার করতে পারেন, তবে সেগুলো অবশ্যই পুরোনো থেকে নতুন ক্রমে পাঠান যাতে ক্লায়েন্ট ক্রমানুসারে স্টেট পুনর্গঠন করতে পারে।

ডুপ্লিকেট মেসেজের জন্য প্রস্তুত থাকুন

নেটওয়ার্ক সবসময় নির্ভরযোগ্য নয়। সার্ভার একটি ইভেন্ট পাঠাতে পারে, কিন্তু TCP অ্যাকনলেজমেন্ট হারিয়ে ফেলতে পারে এবং টাইমআউটের পরে সেটি আবার পাঠাতে পারে। শুরু থেকেই 'at-least-once delivery' মডেল অনুযায়ী ডিজাইন করুন।

ক্লায়েন্ট সাইডে ডুপ্লিকেট রিমুভ করা (deduplication) সহজ। ইভেন্ট ID দিয়ে একটি Map রাখুন। যখন একটি নতুন ইভেন্ট আসবে, ম্যাপটি চেক করুন। যদি ID-টি আগে থেকেই থাকে, তবে ডুপ্লিকেটটি নিঃশব্দে বাদ দিন। যেহেতু আপনার সার্ভার নির্দিষ্ট বা deterministic ID প্রদান করে, তাই ডুপ্লিকেটগুলো ক্ষতিকারক হবে না। ম্যাপটি চিরকাল বড় করার প্রয়োজন নেই। একবার নিশ্চিত হয়ে নিন যে একটি ইভেন্ট নিরাপদে প্রসেস করা হয়েছে, তারপর পুরোনো ID গুলো মুছে ফেলুন। ব্রাউজার ক্লায়েন্টের জন্য কয়েকশ এন্ট্রির একটি স্লাইডিং উইন্ডো (sliding window) সাধারণত যথেষ্ট।

যখন কার্সার এক্সপায়ার হয়ে যায়

অবশেষে কোনো ক্লায়েন্ট কয়েক ঘণ্টা বা কয়েক দিন পর পুনরায় কানেক্ট করতে পারে। যদি আপনার হিস্ট্রি বাফার শুধুমাত্র শেষ এক হাজার ইভেন্ট ধারণ করে এবং ক্লায়েন্টটি দুই হাজার ইভেন্ট পেছনে থাকে, তবে গ্যাপগুলো প্লে করা অসম্ভব হয়ে পড়বে।

আংশিক হিস্ট্রি স্ট্রিম করবেন না। এতে ক্লায়েন্ট একটি অসংলগ্ন (inconsistent) অবস্থায় পড়ে যাবে। পরিবর্তে, একটি এক্সপায়ার হওয়া কার্সার শনাক্ত করুন এবং পরবর্তী ইভেন্ট হিসেবে একটি ফুল স্ন্যাপশট (full snapshot) পাঠান। স্ন্যাপশটটিতে একটি নতুন কার্সার থাকা উচিত যা ক্লায়েন্টকে বর্তমান অবস্থার সাথে যুক্ত করবে। সেখান থেকে, লাইভ ডেল্টা (live deltas) স্বাভাবিকভাবে চলতে থাকবে। আপনার প্রোটোকলে এই সীমানাটি স্পষ্টভাবে উল্লেখ করুন যাতে ক্লায়েন্ট কোড জানে কখন তার লোকাল মডেলটি অ্যাপেন্ড (append) করার পরিবর্তে রিসেট করতে হবে।

স্ট্রিম সুরক্ষিত রাখুন

ওপেন SSE এন্ডপয়েন্টগুলো আক্রমণকারীদের জন্য আকর্ষণীয় লক্ষ্য হতে পারে। যে কেউ একটি কানেকশন ধরে রাখতে পারে এবং রিকয়েস্ট রিপ্লে করার মাধ্যমে আপনার স্টোরেজের ওপর রিড লোড (read load) বাড়িয়ে দিতে পারে।

সঠিক অথরাইজেশন দিয়ে এন্ডপয়েন্টটি সুরক্ষিত করুন। যেহেতু ব্রাউজারের EventSource কাস্টম হেডার সাপোর্ট করে না, তাই টোকেনটি কুয়েরি স্ট্রিং-এ পাস করুন অথবা কঠোর SameSite পলিসি সহ কুকি ব্যবহার করুন। স্ট্রীম রিসোর্স বরাদ্দ করার আগে টোকেনটি যাচাই করে নিন।

হিস্ট্রি লিমিট এবং প্রতি-ব্যবহারকারীর কোটা সেট করুন। প্রতিটি টাস্কের জন্য সংরক্ষিত ইভেন্টের সংখ্যা এবং প্রতিটি ক্লায়েন্টের জন্য সমসাময়িক (concurrent) কানেকশনের সংখ্যা নির্দিষ্ট করে দিন। ডিসকানেক্ট এবং রিপ্লেগুলো লগ করুন যাতে আপনি কোনো ক্ষতিকারক ক্লায়েন্ট (rogue client) আপনার কার্সার এন্ডপয়েন্টে অতিরিক্ত রিকোয়েস্ট পাঠালে তা শনাক্ত করতে পারেন।

এই প্যাটার্নটি সর্বত্র প্রযোজ্য

এই পদ্ধতিটি শুধুমাত্র HTTP-এর মধ্যেই সীমাবদ্ধ নয়। আপনি যখন WebSockets, message queues, বা agent-to-agent ইন্টারফেস ব্যবহার করবেন, তখনও একই নিয়ম প্রযোজ্য হবে। ট্রান্সপোর্ট মাধ্যম পরিবর্তিত হতে পারে—আপনি বাইনারি ফ্রেম বা টপিক সাবস্ক্রিপশন ব্যবহার করতে পারেন—কিন্তু মূল সমস্যাটি একই থাকে। আপনার একটি কার্সার, একটি ডিউরেবল লগ (durable log), at-least-once semantics, ক্লায়েন্ট ডিডুপ্লিকেশন (client deduplication) এবং কার্সারটি পুরোনো হয়ে গেলে ফুল স্ন্যাপশটে (full snapshot) ফিরে যাওয়ার ব্যবস্থা প্রয়োজন। স্টেট কনভারজেন্সের (state convergence) সমাধান একবার করে ফেললে, আপনি মূল লজিক পুনরায় ডিজাইন না করেই এটি TCP, WebSocket, বা RabbitMQ-এর মতো কোনো ব্রোকারের মাধ্যমে পাঠাতে পারবেন।

সহজ রাখুন

Server-Sent Events কাজ করে কারণ এগুলো সাধারণ HTTP-এর ওপর ভিত্তি করে চলে। প্রক্সিগুলো এগুলো বুঝতে পারে। লোড ব্যালেন্সারগুলো এগুলো হেলথ-চেক করতে পারে। ডিবাগিং করা curl-এর মতোই সহজ। কিন্তু আপনি যদি এজ কেসগুলো (edge cases) উপেক্ষা করেন, তবে সেই সহজলভ্যতা হারিয়ে যাবে। কার্সার তৈরি করুন। রিপ্লে (replays) আশা করুন। ক্লায়েন্ট সাইডে ডিডুপ্লিকেশন করুন। হিস্ট্রি শেষ হয়ে গেলে স্ন্যাপশট নিন। এটি করলে, আপনার দীর্ঘমেয়াদী AI টাস্কগুলো দুর্বল Wi-Fi, সার্ভার রিস্টার্ট এবং মাঝেমধ্যে রাতে ব্রাউজার স্লিপ মোডে চলে গেলেও তাদের প্রগ্রেস সঠিকভাবে রিপোর্ট করতে পারবে।

উৎস: Build a Reconnecting SSE Task Stream with Node.js

আলোচনায় যোগ দিন: GyaanSetu AI Community