Design Search Autocomplete

Problem এর পরিচয়

Autocomplete দেখতে ছোট একটা feature: search box এর নিচে একটা dropdown। কিন্তু HLD problem হিসেবে এটা দারুণ, কারণ সংখ্যাগুলো এমন একটা design এ ঠেলে দেয় যেটা "database এ query মারো" থেকে একদম আলাদা: প্রতিটা key চাপা একটা request, পরের অক্ষর টাইপ করার আগেই উত্তর আসতে হবে, আর data (মানুষ কী খোঁজে) সারাদিন বদলায়। মূল শিক্ষাটা অনেক জায়গায় কাজে লাগবে: system টাকে দুই ভাগ করো, একটা ধীর offline পথ যেটা উত্তর বানিয়ে রাখে, আর একটা দ্রুত online পথ যেটা শুধু সেই উত্তর পড়ে।

Problem statement

মিরপুরের কেউ একটা বড় shopping site খুলে "dha" টাইপ করা শুরু করল। শব্দটা শেষ করার আগেই dropdown এ চলে এসেছে "dhaka weather", "dhakaiya biryani", "dhaka traffic"। একটায় tap করল, বাকিটা আর টাইপ করতে হলো না। Phone এর keyboard এ এতে সত্যিই কষ্ট বাঁচে, আর মানুষ এমন query তে যায় যেটার আসলে result আছে।

কী বানাচ্ছি: এমন একটা service, user এ পর্যন্ত যা টাইপ করেছে (prefix) সেটা দিলে ওই prefix দিয়ে শুরু হওয়া সবচেয়ে popular 10 টা পুরো query ফেরত দেয়, এত দ্রুত যে মনে হয় সাথে সাথে চলে এলো।

কে ব্যবহার করে: search box এ যে টাইপ করে সবাই, প্রতিটা key চাপায়। পরোক্ষভাবে search team, যাদের দরকার suggestion যেন fresh থাকে (নতুন metro line, Eid sale, breaking news) আর পরিষ্কার থাকে (কোনো গালি বা কারো personal তথ্য suggest হবে না)।

কী সমস্যা মেটায়: কম টাইপ, কম বানান ভুল, আর এমন search যেটা আসলে আছে এমন জিনিসে গিয়ে পড়ে।

Technically কোথায় মজা:

• Latency: mobile network সহ পুরো আসা-যাওয়া 100 ms এর অনেক কমে শেষ হতে হবে, নাহলে dropdown টাইপিং এর পেছনে পড়ে থাকে, কোনো কাজে আসে না।
• পরিমাণ: একটা search মানে কয়েকটা request (টাইপের প্রতিটা বিরতিতে একটা), তাই suggestion এর traffic search এর traffic এর কয়েক গুণ।
• ছোট prefix বিশাল: "d" দিয়ে শুরু হওয়া query লক্ষ লক্ষ। প্রতিটা key চাপায় এগুলো sort করা অসম্ভব।
• Fresh বনাম দ্রুত: কোনটা popular সেটা দিনের মধ্যেই বদলায়, কিন্তু request এর পথে উত্তর নতুন করে বানালে latency শেষ।
• Content safety: মানুষ যা টাইপ করে সেটাই লাখো মানুষকে suggestion হিসেবে দেখানো হতে পারে, গালি, গুজব, phone number সহ।

Requirement

Functional requirement

1. Suggest: একটা prefix এর জন্য ওটা দিয়ে শুরু হওয়া সর্বোচ্চ 10 টা পুরো query ফেরত দাও, সবচেয়ে popular টা আগে।
2. শুধু আসল search: suggestion আসবে মানুষ আসলে যেগুলো submit করেছে (Enter চেপেছে) সেগুলো থেকে, প্রতিটা key চাপা থেকে না।
3. Freshness: কোনো query popular হয়ে গেলে মোটামুটি এক ঘণ্টার মধ্যে দেখাবে; হঠাৎ ভাইরাল হওয়া trend মোটামুটি 15 মিনিটে (stretch goal)।
4. Filtering: blocked শব্দ, আপত্তিকর কিছু, আর personal data র মতো দেখতে কিছু কখনো suggest হবে না; অভিযোগ এলে কয়েক মিনিটের মধ্যে একটা suggestion সরানো যাবে।
5. Scope এর বাইরে: বানান ঠিক করা ("dhkaa"), personal suggestion, আর search result page নিজে। প্রথম দুইটা interview অংশে আলোচনা করব।

Non-functional requirement

Latency আর throughput ই design ঠিক করে। খেয়াল করো কোনগুলো কড়া না: freshness মিনিটে মাপা, durability ঢিলা, আর availability 99.9% হলেই চলে কারণ feature টা ভাঙলেও search চলে। এই ছাড়টুকুর জন্যই আমরা সবকিছু আগে থেকে হিসাব করে রাখতে পারি।

Capacity আন্দাজ

ধরে নেওয়া (মুখে বলা, গোল সংখ্যা):

• দিনে 5 কোটি active user, প্রত্যেকে দিনে 6 টা search করে।
• একটা query গড়ে 18 অক্ষর, কিন্তু ছোট একটা debounce থাকায় browser প্রতি search এ মোটামুটি 6 টা suggestion request পাঠায়।
• সবচেয়ে ব্যস্ত ঘণ্টা গড়ের মোটামুটি 3 গুণ।
• 30 দিনে অন্তত 20 জন আলাদা মানুষ যেটা search করেছে শুধু সেটাই suggestion হতে পারবে (এটা privacy rule ও, Security অংশ দেখো)। তাতে মোটামুটি 1 কোটি suggest করার মতো query থাকে, গড়ে 20 byte লম্বা।
• প্রতিটা prefix এ top 10 রাখব।

Traffic

Storage

Server: memory তে trie lookup কয়েক microsecond এর কাজ; আসল সময় যায় HTTP এর আশেপাশের কাজে। একটা server সেকেন্ডে 10,000 request সামলায় ধরে নিলে (হিসেব করে কম ধরা), peak এ মোটামুটি 7 টা server লাগে। দুইটা zone মিলিয়ে 10 টা চালাও, যাতে একটা zone পড়ে গেলেও সমস্যা না হয়। প্রতিটা server পুরো 5 GB trie রাখে, তাই যেকোনো server যেকোনো prefix এর উত্তর দিতে পারে।

Bandwidth: 10 টা suggestion সহ একটা response মোটামুটি 400 byte JSON আর ~300 byte header। Peak এ 63,000 × 700 B ≈ 44 MB/s (~350 Mbit/s)। চোখে পড়ার মতো, তবে সমস্যা না, আর gzip আর CDN এটা আরো কমায়।

Cache size: 1 থেকে 3 অক্ষরের prefix মোট মোটামুটি 37 + 37² + 37³ ≈ 52,000 টা (অক্ষর, সংখ্যা, space)। প্রতিটা 700 B হলে 36 MB, আর এই ছোট prefix গুলোই সবচেয়ে বেশি বার আসে, তাই CDN edge cache এর জন্য একদম মাপমতো।

সংখ্যাগুলো কী বলছে: serve করার data (5 GB) একটা machine এর memory তে ধরে যায়, তাই একটা lookup কোনো database ছাড়াই শুধু pointer ধরে হাঁটা হতে পারে। খরুচে কাজটা (30 কোটি search গোনা আর প্রতিটা prefix এর ranking) offline এ, batch এ করা যায়, কারণ freshness মিনিটে মাপা। পুরো design টা এই দুই বাক্যেই।

API design

• GET আর URL এ prefix কেন: এতে browser আর CDN URL ধরে response cache করতে পারে। এই API এর সবচেয়ে জরুরি গুণ এটাই।
• Response এ count নেই: popularity এর সংখ্যা ভেতরের ব্যাপার; বাইরে দেখালে মানুষ সেটা নিয়ে কারসাজি করতে চায়।
• Authentication: suggestion anonymous। ইচ্ছা করেই: request এ cookie বা user ID থাকলে response personal হয়ে যায়, আর cache করা যায় না। Rate limit হয় IP আর client app ধরে।
• Idempotency: GET এমনিতেই idempotent। Search event এ event_id থাকে, যাতে log একটা event দুইবার দিলে গোনার pipeline duplicate বাদ দিতে পারে।
• version বলে কোন trie উত্তর দিয়েছে; "এই suggestion কেন দেখলাম" debug করতে কাজে লাগে, আর পুরনো version এর কোনো response দেরিতে এলে client সেটা বাদ দিতে পারে।
• Client এর দায়িত্ব: browser পাঠানোর আগে একটু থামে (মোটামুটি 50–100 ms), নতুন key চাপলে আগের request cancel করে, আর session জুড়ে উত্তর cache করে। এই তিনটা নিয়মেই server এর traffic অনেক কমে যায়।

Database design

Access pattern: online পথ পড়ে "ঠিক এই prefix এর top 10", সেকেন্ডে হাজার হাজার বার, আর কখনো লেখে না। Offline পথ সারাদিন search event যোগ করে, আর ঘণ্টায় একবার সব count নতুন করে লেখে।

সোজা database এ query করলে কী সমস্যা? SELECT query FROM queries WHERE query LIKE 'd%' ORDER BY count DESC LIMIT 10 prefix এর জন্য B-tree index ব্যবহার করতে পারে, কিন্তু "d" এর জন্য তবুও লক্ষ লক্ষ মিলে যাওয়া row পড়ে count ধরে sort করতে হয়। 1,000 query এর site এ এটা ঠিক আছে, আমাদের scale এ আশা নেই। সমাধান হলো প্রতিটা prefix এর উত্তর আগেই হিসাব করে রাখা, আর সেই উত্তরগুলো ঘন করে রাখার একটা উপায় হলো trie।

তাই serve করার "database" হলো trie নিজেই, memory তে, একটা file থেকে load করা। বাকি store গুলো সাধারণ:

-- Raw search events: an append-only log (Kafka topic), then hourly files in object storage
-- (Parquet, partitioned by hour). Not a SQL table; shown as columns for clarity.
--   event_id, query, lang, region, user_hash, ts

-- Aggregated counts, rewritten by every batch (warehouse table or Parquet files).
CREATE TABLE query_counts (
  query        TEXT    NOT NULL,
  lang         TEXT    NOT NULL,
  score        BIGINT  NOT NULL,          -- decayed count: old score / 2 + new searches
  users_30d    INT     NOT NULL,          -- distinct users, for the privacy threshold
  PRIMARY KEY (lang, query)
);

-- Which trie version each shard should serve (small, PostgreSQL).
CREATE TABLE trie_versions (
  version     BIGINT PRIMARY KEY,
  shard       TEXT   NOT NULL,            -- e.g. 'a-c', 'd-k', 'l-z'
  path        TEXT   NOT NULL,            -- object storage key of the snapshot file
  checksum    TEXT   NOT NULL,
  queries     INT    NOT NULL,
  status      TEXT   NOT NULL,            -- building, validated, live, rolled_back
  built_at    TIMESTAMPTZ NOT NULL
);

-- Blocked words and removed suggestions (small, PostgreSQL, edited by staff).
CREATE TABLE blocked_terms (
  term        TEXT PRIMARY KEY,
  reason      TEXT NOT NULL,
  added_by    BIGINT NOT NULL,
  added_at    TIMESTAMPTZ NOT NULL DEFAULT now()
);

• Log শুধু append হয় আর সময় ধরে ভাগ করা, তাই সেকেন্ডে 10,000 event লেখা সস্তা, আর এক ঘণ্টার data মানে পড়ার জন্য একটা directory।
• Count নতুন করে হিসাব হয়, জায়গায় update হয় না: batch আগের count আর নতুন ঘণ্টা পড়ে পুরো নতুন একটা set লেখে। কোনো row নিয়ে কাড়াকাড়ি হয় না।
• trie_versions হলো control plane: server গুলো এটা poll করে (বা notification পায়), নিজের shard এর file নামায়, checksum মিলিয়ে swap করে। Rollback মানে পুরনো একটা row কে আবার live করে দেওয়া।
• Replication আর partitioning: log replicate করে Kafka আর object storage; trie "replicate" হয় প্রতিটা server এ একই file load করে; shard হলো prefix range (Scaling দেখো)। ছোট PostgreSQL table গুলোর জন্য একটা standby ই যথেষ্ট।

Architecture এর ছবি

Prefix cache নিয়ে একটা কথা: RAM এ trie ধরে হাঁটা এমনিতেই microsecond এর ব্যাপার, তাই সামনে একটা shared Redis বসালে উল্টো একটা network hop বাড়ে। এখানে আমাদের cache হলো process এর ভেতরের একটা LRU, যেটা hot prefix এর হাঁটা আর JSON বানানো বাঁচায়। Shared Redis এর দাম ওঠে পরে, যখন একটা উত্তর কয়েকটা shard থেকে জোড়া লাগাতে হয় বা personal history র সাথে মেশাতে হয়। এটাই caching strategy: আগে browser, তারপর CDN, তারপর process এর ভেতরে, Redis শুধু খরুচে জোড়া লাগানো উত্তরের জন্য।

যেগুলো ইচ্ছা করে রাখিনি: read পথে database, read পথে কোনো write, বা প্রথম দিন থেকে stream processor (ঘণ্টায় একবারের batch দিয়েই freshness এর requirement মেটে)।

একটা request এর জীবন

একটা prefix টাইপ করা (cache miss)

1. User d, h, a টাইপ করে। শেষ key এর পর browser ~100 ms অপেক্ষা করে, তাই একটাই request যায়: GET /v1/suggest?q=dha। dh এর request তখনো রাস্তায় থাকলে সেটা cancel হয়।
2. আগে browser cache, তারপর CDN: আশেপাশের কেউ গত 60 সেকেন্ডে dha চেয়ে থাকলে edge ই উত্তর দেয়, আমাদের server টের ও পায় না।
3. LB → যেকোনো suggest server। Server prefix টা normalise করে (ছোট হাতের অক্ষর, একটা করে space, সর্বোচ্চ 50 অক্ষর; শেষের space রেখে দেয়, কারণ dhaka আর dhaka আলাদা prefix)।
4. Current trie এর pointer একবার পড়ে (version 1), cache এ 1:dha খোঁজে। Miss।
5. Trie তে d → h → a হাঁটে: তিন ধাপ। Top-10 list ওই node এ আগে থেকেই রাখা, খোঁজা বা sort করার কিছু নেই।
6. List টা 1:dha নামে রাখে আর Cache-Control: public, max-age=60 সহ 200 দেয়। Server এর মোট সময়: এক millisecond এর অনেক কম।

একটা search গোনা হচ্ছে (offline পথ)

1. User "dhaka metro rail" এ Enter চাপে। Search service query log এ {event_id, query, lang, user_hash, ts} লেখে। Search নিজে এর জন্য অপেক্ষা করে না।
2. প্রতি ঘণ্টায় aggregator গত ঘণ্টার event পড়ে, duplicate event_id বাদ দেয়, query normalise করে গোনে।
3. আগের score এর সাথে মেলায়: score = old_score / 2 + new_count (exponential decay, যাতে গত সপ্তাহ মিলিয়ে যায় আর নতুন trend সামনে আসতে পারে), blocked শব্দ আর 20 জন আলাদা user এর নিচের query বাদ দেয়।
4. Builder নতুন trie বানায়, নিচ থেকে উপরে প্রতিটা node এর top 10 হিসাব করে, তারপর যাচাই চালায় (size আগের build এর ±20% এর মধ্যে, কিছু "golden" prefix এখনো ঠিকঠাক উত্তর দেয়, কোনো blocked শব্দ নেই)।
5. Snapshot file checksum সহ object storage এ যায়; trie_versions এর একটা row প্রথমে validated, তারপর live হয়।
6. প্রতিটা suggest server file নামায়, পুরনো trie এর পাশে memory তে load করে, আর একটা pointer swap করে। যে request গুলো চলছিল সেগুলো পুরনো trie তেই শেষ হয়; নতুন গুলো নতুনটা পায়। পুরনো cache entry র key তে পুরনো version, তাই সেগুলো আর কখনো serve হয় না, এমনি এমনি বের হয়ে যায়।

Component গুলো গভীরে

প্রতিটা node এ top-K সহ trie

সাধারণ trie তে "কোন query গুলো d দিয়ে শুরু?" এর উত্তর দিতে d এর নিচের পুরো subtree হাঁটতে হয়: লক্ষ লক্ষ node। আমরা request এর সময় এটা করি না। বরং offline এ বানানোর সময় প্রতিটা node তার subtree র সেরা K টা query রেখে দেয়। হিসাবটা হয় নিচ থেকে উপরে: একটা node এর list হলো তার নিজের query (যদি ওখানে কোনো query শেষ হয়) আর children দের list মিলিয়ে, sort করে, K তে কাটা। Children দের list আগেই নিজের subtree র সেরা, তাই merge শুধু K × children টা জিনিস দেখে, পুরো subtree না।

Production এর জন্য memory বাঁচানোর কৌশল: list এ string এর বদলে 4-byte query ID রাখো; একটাই child আছে এমন node এর শিকল জুড়ে দাও (radix tree); একটা নির্দিষ্ট গভীরতা পর্যন্তই list রাখো, তার নিচে ছোট একটা scan; বা finite-state transducer ব্যবহার করো (Lucene এর suggester যেটা ব্যবহার করে), যেটা prefix এর সাথে suffix ও ভাগাভাগি করে।

Immutable snapshot আর atomic swap

Trie কখনো জায়গায় edit হয় না। প্রতিটা build পুরো নতুন একটা trie বানায়; server সেটা current টার পাশে load করে, তারপর একটা মাত্র pointer swap করে। এতে তিনটা জিনিস ফ্রিতে পাই:

• Read পথে কোনো lock নেই: reader কখনো আধা-update হওয়া trie দেখে না, কারণ কেউ ওটা update ই করে না।
• Consistent উত্তর: একটা request শুরুতে একবারই pointer পড়ে, তাই তার পুরো উত্তর এক version থেকে আসে (Go এর TestReadersDuringSwaps test টা -race সহ ঠিক এটাই যাচাই করে)।
• সাথে সাথে rollback: আগের file এখনো object storage এ আছে; ফিরে যাওয়াও ঠিক একই কাজ।

দাম হলো memory: swap এর সময় একটা server অল্প সময়ের জন্য দুইটা trie রাখে (2 × 5 GB)। Server এর size সেই হিসাবে ঠিক করো, নাহলে shard গুলো একটা একটা করে load করো।

Version সহ cache key

Cache key যদি শুধু prefix হতো, তাহলে swap এর পরেও প্রতিটা entry expire না হওয়া পর্যন্ত cache পুরনো list দিতে থাকত। প্রতিটা swap এ সব server জুড়ে সব key মুছে ফেলা ধীর, আর race এর ঝুঁকি আছে। তার বদলে key তে trie এর version থাকে: 412:dha। 413 এ swap হওয়ার পর প্রতিটা lookup 413:... key ব্যবহার করে, তাই পুরনো entry গুলোতে আর কেউ hit করে না, LRU এগুলো বের করে দেয়। CDN এর key তে আমাদের version বসানো যায় না (ও URL ধরে cache করে), তাই সেখানে max-age (60 s) পর্যন্ত পুরনো থাকা মেনে নিই, freshness এর requirement এটা অনুমতি দেয়।

Decay সহ গোনা, আর trend

Batch হিসাব করে score = old_score / 2 + this_hour (production এ অর্ধেক করার হারটা ঠিক করে নেওয়া হয়, যেমন কয়েক দিনের half-life)। Decay না থাকলে গত বছরের "eid ul adha 2025" এই সপ্তাহের খবরকে চিরকাল হারিয়ে দিত। Decay থাকলে হঠাৎ অনেক search হওয়া একটা query এক-দুইটা build এর মধ্যেই পুরনো প্রিয় query গুলোকে ছাড়িয়ে যায়। Demo ঠিক এটাই দেখায়: "dhaka metro rail" 1 টা search থেকে এক দিনের ভারী search এর পর d এর একদম উপরে চলে যায়।

"15 মিনিটে breaking news" এর জন্য ঘণ্টার batch যথেষ্ট না। শেষ design এ ছোট একটা stream job যোগ হয় যেটা গত 15 মিনিটের count রাখে, যে query গুলো স্বাভাবিকের চেয়ে অনেক দ্রুত বাড়ছে সেগুলো খুঁজে বের করে, আর server গুলোতে ছোট একটা "boost list" পাঠায়; পরের batch সেগুলো নিজের মধ্যে নেওয়া পর্যন্ত server গুলো এটা উত্তরে মিশিয়ে দেয়। আলাদা, সহজ একটা পথ, যাতে মূল build টা একঘেয়ে আর নির্ভরযোগ্য থাকে।

চালিয়ে দেখার মতো core

এই program টা design এর মূল অংশ, এক বসায় পড়ে ফেলার মতো ছোট: normalisation, submit করা search গোনার একটা aggregator, decay আর blocklist সহ একটা merge, প্রতিটা prefix এর top K আগে থেকে হিসাব করে রাখা একটা trie, version:prefix key এর একটা LRU cache, আর একটা service যেটা key চাপার উত্তর দেয় আর atomic ভাবে নতুন trie swap করে।

autocomplete/
├── go.mod
├── trie.go               Trie: Build (with top K per node), Top
├── aggregate.go          Normalize, Aggregator, Merge (decay + blocklist)
├── cache.go              LRU cache of version:prefix -> suggestions
├── service.go            Service: Suggest, Publish (atomic swap)
├── autocomplete_test.go  top K vs brute force, swaps under -race, versioned cache, merge
└── cmd/demo/main.go      two days of searches, one user typing, a rebuild and a swap

// trie.go
package autocomplete

import "sort"

// Suggestion is one query and how often people searched for it.
type Suggestion struct {
	Query string
	Count int
}

// better orders suggestions: more searches first, then alphabetically, so every build and
// every language gives the same list.
func better(a, b Suggestion) bool {
	if a.Count != b.Count {
		return a.Count > b.Count
	}
	return a.Query < b.Query
}

type node struct {
	kids  map[byte]*node
	count int          // > 0 when a whole query ends here
	top   []Suggestion // the best k queries that start with this prefix, precomputed
}

// Trie is a read-only snapshot built offline from search counts. Every node already knows
// its top k, so a lookup is just "walk the prefix, return the list": no searching at query time.
type Trie struct {
	root    *node
	k       int
	Version int
	Queries int
	Nodes   int
}

// Build makes a trie from query counts and fills in the top k of every prefix.
func Build(version int, counts map[string]int, k int) *Trie {
	t := &Trie{root: &node{kids: map[byte]*node{}}, k: k, Version: version, Nodes: 1}
	for q, c := range counts {
		if c <= 0 || q == "" {
			continue
		}
		n := t.root
		for i := 0; i < len(q); i++ {
			next, ok := n.kids[q[i]]
			if !ok {
				next = &node{kids: map[byte]*node{}}
				n.kids[q[i]] = next
				t.Nodes++
			}
			n = next
		}
		n.count = c
		t.Queries++
	}
	t.fill(t.root, nil)
	return t
}

// fill computes top k bottom-up: a node's list is its own query plus its children's lists.
func (t *Trie) fill(n *node, path []byte) {
	var all []Suggestion
	if n.count > 0 {
		all = append(all, Suggestion{string(path), n.count})
	}
	for c, kid := range n.kids {
		t.fill(kid, append(path, c))
		all = append(all, kid.top...)
	}
	sort.Slice(all, func(i, j int) bool { return better(all[i], all[j]) })
	if len(all) > t.k {
		all = all[:t.k]
	}
	n.top = all
}

// Top returns the precomputed best k for a prefix (nil if nobody searched it). The slice
// belongs to the snapshot: callers must not change it.
func (t *Trie) Top(prefix string) []Suggestion {
	n := t.root
	for i := 0; i < len(prefix); i++ {
		if n = n.kids[prefix[i]]; n == nil {
			return nil
		}
	}
	return n.top
}

// aggregate.go
package autocomplete

import (
	"strings"
	"sync"
)

// MaxLen is the longest query we store or look up, in bytes.
const MaxLen = 50

// Normalize makes "  Dhaka   WEATHER " and "dhaka weather" the same query: lowercase, single
// spaces, trimmed, at most MaxLen bytes. (This demo uses ASCII; production also normalises
// Unicode, and Bangla queries work the same way.)
func Normalize(s string) string {
	s = strings.Join(strings.Fields(strings.ToLower(s)), " ")
	if len(s) > MaxLen {
		s = strings.TrimRight(s[:MaxLen], " ")
	}
	return s
}

// Aggregator counts searches as they happen, like the log pipeline that feeds the builder.
type Aggregator struct {
	mu     sync.Mutex
	counts map[string]int
	Seen   int
}

func NewAggregator() *Aggregator { return &Aggregator{counts: map[string]int{}} }

// Add records one search that the user actually submitted.
func (a *Aggregator) Add(raw string) {
	q := Normalize(raw)
	if q == "" {
		return
	}
	a.mu.Lock()
	defer a.mu.Unlock()
	a.counts[q]++
	a.Seen++
}

// Drain hands over the counts collected so far and starts a new period.
func (a *Aggregator) Drain() map[string]int {
	a.mu.Lock()
	defer a.mu.Unlock()
	out := a.counts
	a.counts = map[string]int{}
	a.Seen = 0
	return out
}

// Merge prepares the next build: old counts are halved (so last week fades and new trends can
// win), this period's counts are added, and queries with a blocked word are removed.
func Merge(old, recent map[string]int, blocked map[string]bool) (counts map[string]int, removed int) {
	counts = map[string]int{}
	for q, c := range old {
		if c/2 > 0 {
			counts[q] = c / 2
		}
	}
	for q, c := range recent {
		counts[q] += c
	}
	for q := range counts {
		for _, w := range strings.Fields(q) {
			if blocked[w] {
				delete(counts, q)
				removed++
				break
			}
		}
	}
	return counts, removed
}

// cache.go
package autocomplete

import (
	"container/list"
	"sync"
)

// Cache is a small LRU cache of "version:prefix" -> suggestions, like Redis (or a CDN edge)
// in front of the trie servers.
type Cache struct {
	mu           sync.Mutex
	cap          int
	order        *list.List // front = most recently used
	items        map[string]*list.Element
	Hits, Misses int
}

type entry struct {
	key  string
	list []Suggestion
}

func NewCache(capacity int) *Cache {
	return &Cache{cap: capacity, order: list.New(), items: map[string]*list.Element{}}
}

func (c *Cache) Get(key string) ([]Suggestion, bool) {
	c.mu.Lock()
	defer c.mu.Unlock()
	e, ok := c.items[key]
	if !ok {
		c.Misses++
		return nil, false
	}
	c.Hits++
	c.order.MoveToFront(e)
	return e.Value.(entry).list, true
}

func (c *Cache) Put(key string, list []Suggestion) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if e, ok := c.items[key]; ok {
		c.order.MoveToFront(e)
		return
	}
	if c.order.Len() == c.cap {
		old := c.order.Back()
		c.order.Remove(old)
		delete(c.items, old.Value.(entry).key)
	}
	c.items[key] = c.order.PushFront(entry{key, list})
}

// service.go
// Package autocomplete is the core of a search autocomplete service: searches are counted,
// a trie with the top k per prefix is built offline, swapped in atomically, and served
// through a cache whose keys carry the trie's version.
package autocomplete

import (
	"strconv"
	"strings"
	"sync/atomic"
)

// Service is one suggestion server. Lookups never lock the trie: it is read-only, and a new
// build replaces it with one atomic pointer swap.
type Service struct {
	trie    atomic.Pointer[Trie]
	cache   *Cache
	lookups atomic.Int64 // requests that reached the trie (cache misses)
}

func NewService(t *Trie, cache *Cache) *Service {
	s := &Service{cache: cache}
	s.trie.Store(t)
	return s
}

// Publish swaps in a freshly built trie. Requests already running finish on the old one.
func (s *Service) Publish(t *Trie) { s.trie.Store(t) }

// Version is the version of the trie being served now.
func (s *Service) Version() int { return s.trie.Load().Version }

// Lookups is how many requests missed the cache and walked the trie.
func (s *Service) Lookups() int64 { return s.lookups.Load() }

// Suggest answers one keystroke. The prefix is normalised like stored queries, except that a
// trailing space is kept ("dhaka " should not suggest "dhakaiya"). The cache key includes the
// version, so after a swap old answers are never served; they simply age out of the LRU.
func (s *Service) Suggest(raw string) []Suggestion {
	p := Normalize(raw)
	if p == "" {
		return nil
	}
	if strings.HasSuffix(raw, " ") && len(p) < MaxLen {
		p += " "
	}
	t := s.trie.Load() // one snapshot for the whole request
	key := strconv.Itoa(t.Version) + ":" + p
	if list, ok := s.cache.Get(key); ok {
		return list
	}
	s.lookups.Add(1)
	list := t.Top(p)
	s.cache.Put(key, list)
	return list
}

// cmd/demo/main.go
// Command demo counts a day of searches, builds a trie, serves keystrokes through a cache,
// then counts a second day, rebuilds and swaps the new trie in.
package main

import (
	"fmt"
	"strings"

	"autocomplete"
)

func show(s *autocomplete.Service, typed string) {
	list := s.Suggest(typed)
	parts := []string{}
	for _, x := range list {
		parts = append(parts, fmt.Sprintf("%s %d", x.Query, x.Count))
	}
	if len(parts) == 0 {
		parts = append(parts, "(no suggestions)")
	}
	fmt.Printf("  %-12s -> %s\n", `"`+typed+`"`, strings.Join(parts, " | "))
}

func searches(a *autocomplete.Aggregator, q string, times int) {
	for i := 0; i < times; i++ {
		a.Add(q)
	}
}

func main() {
	blocked := map[string]bool{"fake": true}
	agg := autocomplete.NewAggregator()

	// Day 1: what people submitted, messy spelling and all.
	searches(agg, "dhaka weather", 3)
	searches(agg, "Dhaka Weather", 1)
	searches(agg, "  dhaka   weather ", 1)
	searches(agg, "dhakaiya biryani", 4)
	searches(agg, "dhaka fake certificate", 4)
	searches(agg, "dhaka traffic", 3)
	searches(agg, "dhaka to chittagong bus", 2)
	searches(agg, "dal recipe", 2)
	searches(agg, "dhaka metro rail", 1)
	searches(agg, "eid sale", 3)
	searches(agg, "egg price", 1)
	fmt.Printf("day 1: %d searches logged\n", agg.Seen)
	counts, removed := autocomplete.Merge(nil, agg.Drain(), blocked)
	t1 := autocomplete.Build(1, counts, 3)
	fmt.Printf("build v1: %d queries (%d blocked), %d trie nodes, top 3 kept per prefix\n", t1.Queries, removed, t1.Nodes)

	cache := autocomplete.NewCache(8)
	s := autocomplete.NewService(t1, cache)
	fmt.Println("user 1 types \"dhaka m\", one request per keystroke:")
	for _, p := range []string{"d", "dh", "dha", "dhak", "dhaka", "dhaka ", "dhaka m"} {
		show(s, p)
	}
	fmt.Println("user 2 and a few others:")
	for _, p := range []string{"d", "dh", "dha", "  DHAKA  ", "e", "xyz"} {
		show(s, p)
	}
	fmt.Printf("cache: %d hits, %d misses; trie lookups: %d\n", cache.Hits, cache.Misses, s.Lookups())

	// Day 2: a new metro line opens and everyone searches for it.
	searches(agg, "dhaka metro rail", 9)
	searches(agg, "dhaka weather", 1)
	searches(agg, "eid sale", 2)
	fmt.Printf("day 2: %d searches logged\n", agg.Seen)
	counts, _ = autocomplete.Merge(counts, agg.Drain(), blocked)
	t2 := autocomplete.Build(2, counts, 3)
	s.Publish(t2)
	fmt.Printf("build v2: old counts halved + day 2 added -> %d queries; now serving v%d\n", t2.Queries, s.Version())
	for _, p := range []string{"d", "dhaka ", "e"} {
		show(s, p)
	}
	fmt.Printf("cache: %d hits, %d misses; trie lookups: %d\n", cache.Hits, cache.Misses, s.Lookups())
}

Demo চালালে:

$ go run ./cmd/demo
day 1: 25 searches logged
build v1: 8 queries (1 blocked), 83 trie nodes, top 3 kept per prefix
user 1 types "dhaka m", one request per keystroke:
  "d"          -> dhaka weather 5 | dhakaiya biryani 4 | dhaka traffic 3
  "dh"         -> dhaka weather 5 | dhakaiya biryani 4 | dhaka traffic 3
  "dha"        -> dhaka weather 5 | dhakaiya biryani 4 | dhaka traffic 3
  "dhak"       -> dhaka weather 5 | dhakaiya biryani 4 | dhaka traffic 3
  "dhaka"      -> dhaka weather 5 | dhakaiya biryani 4 | dhaka traffic 3
  "dhaka "     -> dhaka weather 5 | dhaka traffic 3 | dhaka to chittagong bus 2
  "dhaka m"    -> dhaka metro rail 1
user 2 and a few others:
  "d"          -> dhaka weather 5 | dhakaiya biryani 4 | dhaka traffic 3
  "dh"         -> dhaka weather 5 | dhakaiya biryani 4 | dhaka traffic 3
  "dha"        -> dhaka weather 5 | dhakaiya biryani 4 | dhaka traffic 3
  "  DHAKA  "  -> dhaka weather 5 | dhaka traffic 3 | dhaka to chittagong bus 2
  "e"          -> eid sale 3 | egg price 1
  "xyz"        -> (no suggestions)
cache: 4 hits, 9 misses; trie lookups: 9
day 2: 12 searches logged
build v2: old counts halved + day 2 added -> 7 queries; now serving v2
  "d"          -> dhaka metro rail 9 | dhaka weather 3 | dhakaiya biryani 2
  "dhaka "     -> dhaka metro rail 9 | dhaka weather 3 | dhaka to chittagong bus 1
  "e"          -> eid sale 3
cache: 4 hits, 12 misses; trie lookups: 12

Output টা পড়া:

• প্রথম দিন আর build: 25 টা search log হয়েছে, কিন্তু "Dhaka Weather" আর " dhaka weather " normalise হয়ে "dhaka weather" হয়ে গেছে, তাই এগুলো 5 টা search সহ একটাই query। "dhaka fake certificate" 4 বার search হয়েছিল, dhaka এর নিচে দ্বিতীয় suggestion হয়ে যেত; build এর আগেই blocklist এটা বাদ দিয়েছে। 8 টা query থেকে 83 node এর trie।
• "dhaka m" টাইপ করা: d থেকে dhaka পর্যন্ত সবগুলোতে একই তিনটা suggestion: "dhaka weather" (5), "dhakaiya biryani" (4), "dhaka traffic" (3)। এটাই ঠিক: এই সব prefix এর নিচে এগুলোই সেরা query। Space টাইপ করার সাথে সাথে dhaka থেকে "dhakaiya biryani" বাদ পড়ে (এটা "dhaka " দিয়ে শুরু না), "dhaka to chittagong bus" ঢুকে যায়। dhaka m এ থাকে শুধু "dhaka metro rail"।
• দ্বিতীয় user: d, dh, dha cache hit: প্রথম user আগেই এগুলো চেয়েছিল। DHAKA normalise হয়ে dhaka (শেষের space রাখা হয়) হয়, সেটাও hit। xyz কিছু দেয় না, আর এই খালি উত্তরটাও cache হয়, তাই কোনো bot আজেবাজে prefix দিয়ে trie কে চাপ দিতে পারে না।
• Cache এর সংখ্যা: এ পর্যন্ত 4 টা hit আর 9 টা miss; শুধু miss গুলোই trie তে হেঁটেছে।
• দ্বিতীয় দিন: নতুন metro line চালু হলো, "dhaka metro rail" 9 বার search হলো। Merge প্রথম দিনকে অর্ধেক করে (weather 5 → 2, সাথে নতুন 1 = 3) আর দ্বিতীয় দিন যোগ করে। "egg price" (1 → 0) হারিয়ে যায়। v2 তে "dhaka metro rail" (9) d আর dhaka দুই জায়গাতেই সবার উপরে।
• Swap এর পর: d, dhaka আর e এক মিনিট আগে cache এ থাকলেও miss, কারণ key এখন 2:d, 1:d না। কেউ পুরনো list দেখে না, আর কাউকে কিছু মুছতেও হয়নি।

Test গুলো সবচেয়ে জরুরি জিনিসগুলো যাচাই করে: random data তে আগে থেকে হিসাব করা top K আর brute-force উত্তর একই, normalisation আর শেষের space, merge এর decay আর blocklist, swap এর পর কখনো পুরনো cache করা list আসে না, 49 টা swap চলার সময় যে reader গুলো চলছে তারা সবসময় ঠিক একটা version থেকে উত্তর পায় (-race সহ চালানো), আর একসাথে গোনায় কোনো search হারায় না। Atomic pointer এর জায়গায় সাধারণ pointer বসালে swap এর test এ -race data race ধরে, তাই test টা সত্যিই swap কে পাহারা দেয়।

$ go test -race -v ./...
=== RUN   TestTopKMatchesBruteForce
--- PASS: TestTopKMatchesBruteForce (0.00s)
=== RUN   TestNormalize
--- PASS: TestNormalize (0.00s)
=== RUN   TestMergeDecaysAndBlocks
--- PASS: TestMergeDecaysAndBlocks (0.00s)
=== RUN   TestSwapNeverServesOldCacheEntries
--- PASS: TestSwapNeverServesOldCacheEntries (0.00s)
=== RUN   TestTrailingSpaceMatters
--- PASS: TestTrailingSpaceMatters (0.00s)
=== RUN   TestReadersDuringSwaps
--- PASS: TestReadersDuringSwaps (0.00s)
=== RUN   TestAggregatorConcurrentAdds
--- PASS: TestAggregatorConcurrentAdds (0.01s)
PASS
ok  	autocomplete	1.022s
?   	autocomplete/cmd/demo	[no test files]

একই core আরো পাঁচটা language এ

Single-threaded version, একই class, একই output। Swap এখনো একটা reference এর assignment; বদলায় শুধু thread থাকলে প্রতিটা language কীভাবে সেটা নিরাপদ রাখে:

• Python: GIL এর নিচে self.trie = new_trie assignment atomic, আর একটা request নিজের reference t ধরে রাখে, তাই ওই request শেষ হওয়া পর্যন্ত পুরনো trie বেঁচে থাকে।
• TypeScript: Node সব request একটা event loop এ চালায়, তাই swap কোনো lookup এর মাঝখানে ঢুকতে পারে না। (Class টার নাম SuggestCache, কারণ Cache browser এর একটা global।)
• Rust: service একটা Rc<Trie> রাখে, প্রতিটা request সেটা clone করে। Thread থাকলে atomic swap এর পেছনে Arc<Trie> লাগবে (যেমন arc-swap crate), আর thread-safe না এমন version compiler নিজেই মানবে না।
• Java: AtomicReference<Trie> হলো Go এর atomic.Pointer এর সরাসরি জোড়া, আর access order এ LinkedHashMap একটা তৈরি LRU।
• C++: service একটা std::shared_ptr<const Trie> রাখে; প্রতিটা request সেটা copy করে, তাতে পুরনো trie বেঁচে থাকে। Thread থাকলে swap এর জন্য std::atomic<std::shared_ptr<const Trie>> (C++20)।

"""Search autocomplete's core: searches are counted, a trie with the top k per prefix is built
offline, swapped in as a whole, and served through a cache whose keys carry the trie's version."""
from collections import OrderedDict
from dataclasses import dataclass, field

MAX_LEN = 50


@dataclass(frozen=True)
class Suggestion:
    query: str
    count: int


def rank(s: Suggestion) -> tuple[int, str]:
    """More searches first, then alphabetical: the same order in every language."""
    return (-s.count, s.query)


@dataclass
class Node:
    kids: dict[str, "Node"] = field(default_factory=dict)
    count: int = 0  # > 0 when a whole query ends here
    top: list[Suggestion] = field(default_factory=list)  # best k with this prefix, precomputed


class Trie:
    """A read-only snapshot built offline. Every node already knows its top k."""

    def __init__(self, version: int, counts: dict[str, int], k: int):
        self.root, self.k, self.version = Node(), k, version
        self.queries, self.nodes = 0, 1
        for q, c in counts.items():
            if c <= 0 or not q:
                continue
            n = self.root
            for ch in q:
                if ch not in n.kids:
                    n.kids[ch] = Node()
                    self.nodes += 1
                n = n.kids[ch]
            n.count = c
            self.queries += 1
        self._fill(self.root, "")

    def _fill(self, n: Node, path: str) -> None:
        """Bottom-up: a node's list is its own query plus its children's lists."""
        everything = [Suggestion(path, n.count)] if n.count > 0 else []
        for ch, kid in n.kids.items():
            self._fill(kid, path + ch)
            everything += kid.top
        n.top = sorted(everything, key=rank)[: self.k]

    def top(self, prefix: str) -> list[Suggestion]:
        n = self.root
        for ch in prefix:
            n = n.kids.get(ch)
            if n is None:
                return []
        return n.top


def normalize(s: str) -> str:
    """Lowercase, single spaces, trimmed, at most MAX_LEN characters."""
    s = " ".join(s.lower().split())
    return s[:MAX_LEN].rstrip(" ")


class Aggregator:
    """Counts submitted searches, like the log pipeline that feeds the builder."""

    def __init__(self):
        self.counts: dict[str, int] = {}
        self.seen = 0

    def add(self, raw: str) -> None:
        q = normalize(raw)
        if q:
            self.counts[q] = self.counts.get(q, 0) + 1
            self.seen += 1

    def drain(self) -> dict[str, int]:
        out, self.counts, self.seen = self.counts, {}, 0
        return out


def merge(old: dict[str, int], recent: dict[str, int], blocked: set[str]) -> tuple[dict[str, int], int]:
    """Old counts halved, recent counts added, queries with a blocked word removed."""
    counts = {q: c // 2 for q, c in old.items() if c // 2 > 0}
    for q, c in recent.items():
        counts[q] = counts.get(q, 0) + c
    kept = {q: c for q, c in counts.items() if not any(w in blocked for w in q.split())}
    return kept, len(counts) - len(kept)


class Cache:
    """LRU cache of "version:prefix" -> suggestions, like Redis in front of the trie servers."""

    def __init__(self, capacity: int):
        self.cap, self.items = capacity, OrderedDict()
        self.hits = self.misses = 0

    def get(self, key: str) -> list[Suggestion] | None:
        if key not in self.items:
            self.misses += 1
            return None
        self.hits += 1
        self.items.move_to_end(key)
        return self.items[key]

    def put(self, key: str, value: list[Suggestion]) -> None:
        if key in self.items:
            self.items.move_to_end(key)
            return
        if len(self.items) == self.cap:
            self.items.popitem(last=False)
        self.items[key] = value


class Service:
    """One suggestion server. A new build replaces the trie in one assignment."""

    def __init__(self, trie: Trie, cache: Cache):
        self.trie, self.cache, self.lookups = trie, cache, 0

    def publish(self, trie: Trie) -> None:
        self.trie = trie

    def suggest(self, raw: str) -> list[Suggestion]:
        p = normalize(raw)
        if not p:
            return []
        if raw.endswith(" ") and len(p) < MAX_LEN:
            p += " "  # "dhaka " must not suggest "dhakaiya"
        t = self.trie  # one snapshot for the whole request
        key = f"{t.version}:{p}"
        cached = self.cache.get(key)
        if cached is not None:
            return cached
        self.lookups += 1
        found = t.top(p)
        self.cache.put(key, found)
        return found


def show(s: Service, typed: str) -> None:
    parts = [f"{x.query} {x.count}" for x in s.suggest(typed)] or ["(no suggestions)"]
    label = '"' + typed + '"'
    print(f"  {label:<12} -> {' | '.join(parts)}")


def searches(a: Aggregator, q: str, times: int) -> None:
    for _ in range(times):
        a.add(q)


def main() -> None:
    blocked = {"fake"}
    agg = Aggregator()
    # Day 1: what people submitted, messy spelling and all.
    searches(agg, "dhaka weather", 3)
    searches(agg, "Dhaka Weather", 1)
    searches(agg, "  dhaka   weather ", 1)
    searches(agg, "dhakaiya biryani", 4)
    searches(agg, "dhaka fake certificate", 4)
    searches(agg, "dhaka traffic", 3)
    searches(agg, "dhaka to chittagong bus", 2)
    searches(agg, "dal recipe", 2)
    searches(agg, "dhaka metro rail", 1)
    searches(agg, "eid sale", 3)
    searches(agg, "egg price", 1)
    print(f"day 1: {agg.seen} searches logged")
    counts, removed = merge({}, agg.drain(), blocked)
    t1 = Trie(1, counts, 3)
    print(f"build v1: {t1.queries} queries ({removed} blocked), {t1.nodes} trie nodes, top 3 kept per prefix")

    cache = Cache(8)
    s = Service(t1, cache)
    print('user 1 types "dhaka m", one request per keystroke:')
    for p in ["d", "dh", "dha", "dhak", "dhaka", "dhaka ", "dhaka m"]:
        show(s, p)
    print("user 2 and a few others:")
    for p in ["d", "dh", "dha", "  DHAKA  ", "e", "xyz"]:
        show(s, p)
    print(f"cache: {cache.hits} hits, {cache.misses} misses; trie lookups: {s.lookups}")

    # Day 2: a new metro line opens and everyone searches for it.
    searches(agg, "dhaka metro rail", 9)
    searches(agg, "dhaka weather", 1)
    searches(agg, "eid sale", 2)
    print(f"day 2: {agg.seen} searches logged")
    counts, _ = merge(counts, agg.drain(), blocked)
    t2 = Trie(2, counts, 3)
    s.publish(t2)
    print(f"build v2: old counts halved + day 2 added -> {t2.queries} queries; now serving v{s.trie.version}")
    for p in ["d", "dhaka ", "e"]:
        show(s, p)
    print(f"cache: {cache.hits} hits, {cache.misses} misses; trie lookups: {s.lookups}")


if __name__ == "__main__":
    main()

// Search autocomplete's core: searches are counted, a trie with the top k per prefix is built
// offline, swapped in as a whole, and served through a cache whose keys carry the trie's version.

const MAX_LEN = 50;

type Suggestion = { query: string; count: number };

/** More searches first, then alphabetical: the same order in every language. */
function better(a: Suggestion, b: Suggestion): number {
  if (a.count !== b.count) return b.count - a.count;
  return a.query < b.query ? -1 : a.query > b.query ? 1 : 0;
}

class TrieNode {
  kids = new Map<string, TrieNode>();
  count = 0; // > 0 when a whole query ends here
  top: Suggestion[] = []; // best k with this prefix, precomputed
}

/** A read-only snapshot built offline. Every node already knows its top k. */
class Trie {
  root = new TrieNode();
  queries = 0;
  nodes = 1;
  readonly version: number;
  readonly k: number;

  constructor(version: number, counts: Map<string, number>, k: number) {
    this.version = version;
    this.k = k;
    for (const [q, c] of counts) {
      if (c <= 0 || q === "") continue;
      let n = this.root;
      for (const ch of q) {
        let next = n.kids.get(ch);
        if (!next) {
          next = new TrieNode();
          n.kids.set(ch, next);
          this.nodes++;
        }
        n = next;
      }
      n.count = c;
      this.queries++;
    }
    this.fill(this.root, "");
  }

  /** Bottom-up: a node's list is its own query plus its children's lists. */
  private fill(n: TrieNode, path: string): void {
    const all: Suggestion[] = n.count > 0 ? [{ query: path, count: n.count }] : [];
    for (const [ch, kid] of n.kids) {
      this.fill(kid, path + ch);
      all.push(...kid.top);
    }
    n.top = all.sort(better).slice(0, this.k);
  }

  top(prefix: string): Suggestion[] {
    let n: TrieNode | undefined = this.root;
    for (const ch of prefix) {
      n = n.kids.get(ch);
      if (!n) return [];
    }
    return n.top;
  }
}

/** Lowercase, single spaces, trimmed, at most MAX_LEN characters. */
function normalize(s: string): string {
  const one = s.toLowerCase().split(/\s+/).filter((w) => w !== "").join(" ");
  return one.slice(0, MAX_LEN).replace(/ +$/, "");
}

/** Counts submitted searches, like the log pipeline that feeds the builder. */
class Aggregator {
  counts = new Map<string, number>();
  seen = 0;

  add(raw: string): void {
    const q = normalize(raw);
    if (q === "") return;
    this.counts.set(q, (this.counts.get(q) ?? 0) + 1);
    this.seen++;
  }

  drain(): Map<string, number> {
    const out = this.counts;
    this.counts = new Map();
    this.seen = 0;
    return out;
  }
}

/** Old counts halved, recent counts added, queries with a blocked word removed. */
function merge(old: Map<string, number>, recent: Map<string, number>, blocked: Set<string>): [Map<string, number>, number] {
  const counts = new Map<string, number>();
  for (const [q, c] of old) if (Math.floor(c / 2) > 0) counts.set(q, Math.floor(c / 2));
  for (const [q, c] of recent) counts.set(q, (counts.get(q) ?? 0) + c);
  let removed = 0;
  for (const q of [...counts.keys()]) {
    if (q.split(" ").some((w) => blocked.has(w))) {
      counts.delete(q);
      removed++;
    }
  }
  return [counts, removed];
}

/** LRU cache of "version:prefix" -> suggestions, like Redis in front of the trie servers.
 *  (Not called Cache: the DOM already has a global with that name.) */
class SuggestCache {
  items = new Map<string, Suggestion[]>(); // insertion order = least recently used first
  hits = 0;
  misses = 0;
  readonly cap: number;

  constructor(capacity: number) {
    this.cap = capacity;
  }

  get(key: string): Suggestion[] | undefined {
    const v = this.items.get(key);
    if (v === undefined) {
      this.misses++;
      return undefined;
    }
    this.hits++;
    this.items.delete(key);
    this.items.set(key, v);
    return v;
  }

  put(key: string, value: Suggestion[]): void {
    if (this.items.has(key)) {
      this.items.delete(key);
      this.items.set(key, value);
      return;
    }
    if (this.items.size === this.cap) this.items.delete(this.items.keys().next().value!);
    this.items.set(key, value);
  }
}

/** One suggestion server. A new build replaces the trie in one assignment. */
class Service {
  trie: Trie;
  lookups = 0;
  readonly cache: SuggestCache;

  constructor(trie: Trie, cache: SuggestCache) {
    this.trie = trie;
    this.cache = cache;
  }

  publish(trie: Trie): void {
    this.trie = trie;
  }

  suggest(raw: string): Suggestion[] {
    let p = normalize(raw);
    if (p === "") return [];
    if (raw.endsWith(" ") && p.length < MAX_LEN) p += " "; // "dhaka " must not suggest "dhakaiya"
    const t = this.trie; // one snapshot for the whole request
    const key = `${t.version}:${p}`;
    const cached = this.cache.get(key);
    if (cached !== undefined) return cached;
    this.lookups++;
    const found = t.top(p);
    this.cache.put(key, found);
    return found;
  }
}

function show(s: Service, typed: string): void {
  const parts = s.suggest(typed).map((x) => `${x.query} ${x.count}`);
  if (parts.length === 0) parts.push("(no suggestions)");
  console.log(`  ${`"${typed}"`.padEnd(12)} -> ${parts.join(" | ")}`);
}

function searches(a: Aggregator, q: string, times: number): void {
  for (let i = 0; i < times; i++) a.add(q);
}

const blocked = new Set(["fake"]);
const agg = new Aggregator();
// Day 1: what people submitted, messy spelling and all.
searches(agg, "dhaka weather", 3);
searches(agg, "Dhaka Weather", 1);
searches(agg, "  dhaka   weather ", 1);
searches(agg, "dhakaiya biryani", 4);
searches(agg, "dhaka fake certificate", 4);
searches(agg, "dhaka traffic", 3);
searches(agg, "dhaka to chittagong bus", 2);
searches(agg, "dal recipe", 2);
searches(agg, "dhaka metro rail", 1);
searches(agg, "eid sale", 3);
searches(agg, "egg price", 1);
console.log(`day 1: ${agg.seen} searches logged`);
let [counts, removed] = merge(new Map(), agg.drain(), blocked);
const t1 = new Trie(1, counts, 3);
console.log(`build v1: ${t1.queries} queries (${removed} blocked), ${t1.nodes} trie nodes, top 3 kept per prefix`);

const cache = new SuggestCache(8);
const s = new Service(t1, cache);
console.log('user 1 types "dhaka m", one request per keystroke:');
for (const p of ["d", "dh", "dha", "dhak", "dhaka", "dhaka ", "dhaka m"]) show(s, p);
console.log("user 2 and a few others:");
for (const p of ["d", "dh", "dha", "  DHAKA  ", "e", "xyz"]) show(s, p);
console.log(`cache: ${cache.hits} hits, ${cache.misses} misses; trie lookups: ${s.lookups}`);

// Day 2: a new metro line opens and everyone searches for it.
searches(agg, "dhaka metro rail", 9);
searches(agg, "dhaka weather", 1);
searches(agg, "eid sale", 2);
console.log(`day 2: ${agg.seen} searches logged`);
[counts] = merge(counts, agg.drain(), blocked);
const t2 = new Trie(2, counts, 3);
s.publish(t2);
console.log(`build v2: old counts halved + day 2 added -> ${t2.queries} queries; now serving v${s.trie.version}`);
for (const p of ["d", "dhaka ", "e"]) show(s, p);
console.log(`cache: ${cache.hits} hits, ${cache.misses} misses; trie lookups: ${s.lookups}`);

// Search autocomplete's core: searches are counted, a trie with the top k per prefix is built
// offline, swapped in as a whole, and served through a cache whose keys carry the trie's version.
use std::collections::{HashMap, HashSet, VecDeque};
use std::rc::Rc;

const MAX_LEN: usize = 50;

#[derive(Clone, Debug, PartialEq)]
struct Suggestion {
    query: String,
    count: u32,
}

/// More searches first, then alphabetical: the same order in every language.
fn better(a: &Suggestion, b: &Suggestion) -> std::cmp::Ordering {
    b.count.cmp(&a.count).then_with(|| a.query.cmp(&b.query))
}

#[derive(Default)]
struct Node {
    kids: HashMap<u8, Node>,
    count: u32,            // > 0 when a whole query ends here
    top: Vec<Suggestion>,  // best k with this prefix, precomputed
}

/// A read-only snapshot built offline. Every node already knows its top k.
struct Trie {
    root: Node,
    version: u32,
    queries: usize,
    nodes: usize,
}

impl Trie {
    fn build(version: u32, counts: &HashMap<String, u32>, k: usize) -> Trie {
        let mut t = Trie { root: Node::default(), version, queries: 0, nodes: 1 };
        for (q, &c) in counts {
            if c == 0 || q.is_empty() {
                continue;
            }
            let mut n = &mut t.root;
            for &b in q.as_bytes() {
                if !n.kids.contains_key(&b) {
                    t.nodes += 1;
                }
                n = n.kids.entry(b).or_default();
            }
            n.count = c;
            t.queries += 1;
        }
        fill(&mut t.root, &mut Vec::new(), k);
        t
    }

    fn top(&self, prefix: &str) -> &[Suggestion] {
        let mut n = &self.root;
        for b in prefix.as_bytes() {
            match n.kids.get(b) {
                Some(next) => n = next,
                None => return &[],
            }
        }
        &n.top
    }
}

/// Bottom-up: a node's list is its own query plus its children's lists.
fn fill(n: &mut Node, path: &mut Vec<u8>, k: usize) {
    let mut all = Vec::new();
    if n.count > 0 {
        all.push(Suggestion { query: String::from_utf8(path.clone()).unwrap(), count: n.count });
    }
    for (&b, kid) in n.kids.iter_mut() {
        path.push(b);
        fill(kid, path, k);
        path.pop();
        all.extend(kid.top.iter().cloned());
    }
    all.sort_by(better);
    all.truncate(k);
    n.top = all;
}

/// Lowercase, single spaces, trimmed, at most MAX_LEN bytes (ASCII in this demo).
fn normalize(s: &str) -> String {
    let one = s.to_lowercase().split_whitespace().collect::<Vec<_>>().join(" ");
    one[..one.len().min(MAX_LEN)].trim_end_matches(' ').to_string()
}

/// Counts submitted searches, like the log pipeline that feeds the builder.
#[derive(Default)]
struct Aggregator {
    counts: HashMap<String, u32>,
    seen: usize,
}

impl Aggregator {
    fn add(&mut self, raw: &str) {
        let q = normalize(raw);
        if !q.is_empty() {
            *self.counts.entry(q).or_insert(0) += 1;
            self.seen += 1;
        }
    }

    fn drain(&mut self) -> HashMap<String, u32> {
        self.seen = 0;
        std::mem::take(&mut self.counts)
    }
}

/// Old counts halved, recent counts added, queries with a blocked word removed.
fn merge(old: &HashMap<String, u32>, recent: HashMap<String, u32>, blocked: &HashSet<&str>) -> (HashMap<String, u32>, usize) {
    let mut counts: HashMap<String, u32> = old.iter().filter(|(_, &c)| c / 2 > 0).map(|(q, &c)| (q.clone(), c / 2)).collect();
    for (q, c) in recent {
        *counts.entry(q).or_insert(0) += c;
    }
    let before = counts.len();
    counts.retain(|q, _| !q.split(' ').any(|w| blocked.contains(w)));
    let removed = before - counts.len();
    (counts, removed)
}

/// LRU cache of "version:prefix" -> suggestions, like Redis in front of the trie servers.
struct Cache {
    cap: usize,
    items: HashMap<String, Vec<Suggestion>>,
    order: VecDeque<String>, // front = least recently used (fine for a tiny demo cache)
    hits: usize,
    misses: usize,
}

impl Cache {
    fn new(cap: usize) -> Cache {
        Cache { cap, items: HashMap::new(), order: VecDeque::new(), hits: 0, misses: 0 }
    }

    fn touch(&mut self, key: &str) {
        self.order.retain(|k| k != key);
        self.order.push_back(key.to_string());
    }

    fn get(&mut self, key: &str) -> Option<Vec<Suggestion>> {
        match self.items.get(key).cloned() {
            None => {
                self.misses += 1;
                None
            }
            Some(v) => {
                self.hits += 1;
                self.touch(key);
                Some(v)
            }
        }
    }

    fn put(&mut self, key: &str, value: Vec<Suggestion>) {
        if !self.items.contains_key(key) && self.items.len() == self.cap {
            if let Some(old) = self.order.pop_front() {
                self.items.remove(&old);
            }
        }
        self.items.entry(key.to_string()).or_insert(value);
        self.touch(key);
    }
}

/// One suggestion server. A new build replaces the trie by swapping one shared pointer;
/// a request holds its own Rc, so it finishes on the trie it started with.
struct Service {
    trie: Rc<Trie>,
    cache: Cache,
    lookups: usize,
}

impl Service {
    fn publish(&mut self, t: Trie) {
        self.trie = Rc::new(t);
    }

    fn suggest(&mut self, raw: &str) -> Vec<Suggestion> {
        let mut p = normalize(raw);
        if p.is_empty() {
            return Vec::new();
        }
        if raw.ends_with(' ') && p.len() < MAX_LEN {
            p.push(' '); // "dhaka " must not suggest "dhakaiya"
        }
        let t = Rc::clone(&self.trie); // one snapshot for the whole request
        let key = format!("{}:{}", t.version, p);
        if let Some(v) = self.cache.get(&key) {
            return v;
        }
        self.lookups += 1;
        let found = t.top(&p).to_vec();
        self.cache.put(&key, found.clone());
        found
    }
}

fn show(s: &mut Service, typed: &str) {
    let mut parts: Vec<String> = s.suggest(typed).iter().map(|x| format!("{} {}", x.query, x.count)).collect();
    if parts.is_empty() {
        parts.push("(no suggestions)".to_string());
    }
    println!("  {:<12} -> {}", format!("\"{}\"", typed), parts.join(" | "));
}

fn searches(a: &mut Aggregator, q: &str, times: u32) {
    for _ in 0..times {
        a.add(q);
    }
}

fn main() {
    let blocked: HashSet<&str> = ["fake"].into_iter().collect();
    let mut agg = Aggregator::default();
    // Day 1: what people submitted, messy spelling and all.
    searches(&mut agg, "dhaka weather", 3);
    searches(&mut agg, "Dhaka Weather", 1);
    searches(&mut agg, "  dhaka   weather ", 1);
    searches(&mut agg, "dhakaiya biryani", 4);
    searches(&mut agg, "dhaka fake certificate", 4);
    searches(&mut agg, "dhaka traffic", 3);
    searches(&mut agg, "dhaka to chittagong bus", 2);
    searches(&mut agg, "dal recipe", 2);
    searches(&mut agg, "dhaka metro rail", 1);
    searches(&mut agg, "eid sale", 3);
    searches(&mut agg, "egg price", 1);
    println!("day 1: {} searches logged", agg.seen);
    let (counts, removed) = merge(&HashMap::new(), agg.drain(), &blocked);
    let t1 = Trie::build(1, &counts, 3);
    println!("build v1: {} queries ({} blocked), {} trie nodes, top 3 kept per prefix", t1.queries, removed, t1.nodes);

    let mut s = Service { trie: Rc::new(t1), cache: Cache::new(8), lookups: 0 };
    println!("user 1 types \"dhaka m\", one request per keystroke:");
    for p in ["d", "dh", "dha", "dhak", "dhaka", "dhaka ", "dhaka m"] {
        show(&mut s, p);
    }
    println!("user 2 and a few others:");
    for p in ["d", "dh", "dha", "  DHAKA  ", "e", "xyz"] {
        show(&mut s, p);
    }
    println!("cache: {} hits, {} misses; trie lookups: {}", s.cache.hits, s.cache.misses, s.lookups);

    // Day 2: a new metro line opens and everyone searches for it.
    searches(&mut agg, "dhaka metro rail", 9);
    searches(&mut agg, "dhaka weather", 1);
    searches(&mut agg, "eid sale", 2);
    println!("day 2: {} searches logged", agg.seen);
    let (counts, _) = merge(&counts, agg.drain(), &blocked);
    let t2 = Trie::build(2, &counts, 3);
    let queries = t2.queries;
    s.publish(t2);
    println!("build v2: old counts halved + day 2 added -> {} queries; now serving v{}", queries, s.trie.version);
    for p in ["d", "dhaka ", "e"] {
        show(&mut s, p);
    }
    println!("cache: {} hits, {} misses; trie lookups: {}", s.cache.hits, s.cache.misses, s.lookups);
}