Memcached #

SET:
  Request:  set <key> <flags> <exptime> <bytes>\r\n<value>\r\n
  Response: STORED\r\n

GET:
  Request:  get <key>\r\n
  Response: VALUE <key> <flags> <bytes>\r\n<value>\r\n
            END\r\n

DELETE:
  Request:  delete <key>\r\n
  Response: DELETED\r\n  or  NOT_FOUND\r\n

import 'dart:io';
import 'dart:convert';

class MemcachedClient {
  final String host;
  final int port;
  Socket? _socket;
  final _buffer = StringBuffer();

  MemcachedClient({this.host = 'localhost', this.port = 11211});

  Future<void> hubungkan() async {
    _socket = await Socket.connect(host, port);
    _socket!.encoding = utf8;
    print('Connected to Memcached $host:$port');
  }

  Future<void> tutup() async {
    await _socket?.close();
    _socket = null;
  }

  // Send a command and wait for the response
  Future<String> _kirimPerintah(String perintah) async {
    if (_socket == null) throw StateError('Not connected to Memcached');

    _socket!.write(perintah);
    await _socket!.flush();

    // Read the response — wait until a complete line arrives
    final completer = Completer<String>();
    StreamSubscription<List<int>>? sub;

    sub = _socket!.listen((data) {
      final teks = utf8.decode(data);
      _buffer.write(teks);

      final respons = _buffer.toString();
      // Check whether the response is complete
      if (_responLengkap(respons)) {
        sub?.cancel();
        _buffer.clear();
        completer.complete(respons.trim());
      }
    });

    return completer.future.timeout(
      Duration(seconds: 5),
      onTimeout: () => throw TimeoutException('Memcached timeout'),
    );
  }

  bool _responLengkap(String respons) {
    // The response is complete if it ends with END, STORED, DELETED, etc.
    return respons.endsWith('END\r\n') ||
        respons.endsWith('STORED\r\n') ||
        respons.endsWith('NOT_STORED\r\n') ||
        respons.endsWith('DELETED\r\n') ||
        respons.endsWith('NOT_FOUND\r\n') ||
        respons.endsWith('ERROR\r\n') ||
        respons.endsWith('EXISTS\r\n');
  }

  // SET — store a value with a TTL
  // exptime: 0 = never expires, >0 = seconds, >2592000 = Unix timestamp
  Future<bool> set(String key, String value, {int exptime = 0, int flags = 0}) async {
    final bytes = utf8.encode(value).length;
    final perintah = 'set $key $flags $exptime $bytes\r\n$value\r\n';
    final respons = await _kirimPerintah(perintah);
    return respons == 'STORED';
  }

  // GET — fetch a value
  Future<String?> get(String key) async {
    final respons = await _kirimPerintah('get $key\r\n');

    if (respons.startsWith('END')) return null;  // doesn't exist

    // Parse the response: VALUE <key> <flags> <bytes>\r\n<value>\r\nEND
    final baris = respons.split('\r\n');
    if (baris.length < 3) return null;

    return baris[1];  // the second line is the value
  }

  // GETS — GET with a CAS token (for check-and-set)
  Future<({String? value, String? casToken})> gets(String key) async {
    final respons = await _kirimPerintah('gets $key\r\n');

    if (respons.startsWith('END')) return (value: null, casToken: null);

    final baris = respons.split('\r\n');
    final header = baris[0].split(' '); // VALUE <key> <flags> <bytes> <cas>
    return (
      value: baris[1],
      casToken: header.length > 4 ? header[4] : null,
    );
  }

  // DELETE — remove a key
  Future<bool> delete(String key) async {
    final respons = await _kirimPerintah('delete $key\r\n');
    return respons == 'DELETED';
  }

  // ADD — store only if it doesn't exist
  Future<bool> add(String key, String value, {int exptime = 0}) async {
    final bytes = utf8.encode(value).length;
    final respons = await _kirimPerintah(
      'add $key 0 $exptime $bytes\r\n$value\r\n',
    );
    return respons == 'STORED';
  }

  // REPLACE — update only if it already exists
  Future<bool> replace(String key, String value, {int exptime = 0}) async {
    final bytes = utf8.encode(value).length;
    final respons = await _kirimPerintah(
      'replace $key 0 $exptime $bytes\r\n$value\r\n',
    );
    return respons == 'STORED';
  }

  // CAS — Check And Set (update only if the CAS token is still valid)
  Future<bool> cas(String key, String value, String casToken, {int exptime = 0}) async {
    final bytes = utf8.encode(value).length;
    final respons = await _kirimPerintah(
      'cas $key 0 $exptime $bytes $casToken\r\n$value\r\n',
    );
    return respons == 'STORED';
    // STORED = update succeeded
    // EXISTS = someone changed it before us (conflict)
    // NOT_FOUND = the key no longer exists
  }

  // INCR / DECR — increment/decrement numeric values
  Future<int?> incr(String key, int delta) async {
    final respons = await _kirimPerintah('incr $key $delta\r\n');
    if (respons == 'NOT_FOUND') return null;
    return int.tryParse(respons);
  }

  Future<int?> decr(String key, int delta) async {
    final respons = await _kirimPerintah('decr $key $delta\r\n');
    if (respons == 'NOT_FOUND') return null;
    return int.tryParse(respons);
  }

  // STATS — Memcached server info
  Future<Map<String, String>> stats() async {
    final respons = await _kirimPerintah('stats\r\n');
    final map = <String, String>{};
    for (final baris in respons.split('\r\n')) {
      if (baris.startsWith('STAT ')) {
        final bagian = baris.split(' ');
        if (bagian.length >= 3) map[bagian[1]] = bagian[2];
      }
    }
    return map;
  }

  // FLUSH_ALL — delete all data (be careful in production!)
  Future<void> flushAll({int delay = 0}) async {
    await _kirimPerintah('flush_all $delay\r\n');
  }
}

import 'dart:convert';

Future<void> main() async {
  final client = MemcachedClient(host: 'localhost', port: 11211);
  await client.hubungkan();

  try {
    // Basic SET and GET
    await client.set('greeting', 'Hello, Memcached!', exptime: 300);
    final nilai = await client.get('greeting');
    print(nilai); // 'Hello, Memcached!'

    // SET a JSON object
    final produk = {'id': 'P001', 'nama': 'Laptop', 'harga': 15000000};
    await client.set(
      'produk:P001',
      jsonEncode(produk),
      exptime: 3600,  // 1 hour
    );

    final cached = await client.get('produk:P001');
    if (cached != null) {
      final data = jsonDecode(cached) as Map<String, dynamic>;
      print('Product: ${data['nama']}');
    }

    // ADD — only if it doesn't exist
    final sukses = await client.add('kunci_baru', 'nilai', exptime: 60);
    print('Add succeeded: $sukses');

    // INCR for counters
    await client.set('view_count', '0', exptime: 86400);
    final count1 = await client.incr('view_count', 1);
    final count2 = await client.incr('view_count', 1);
    print('View count: $count2'); // 2

    // DELETE
    final dihapus = await client.delete('greeting');
    print('Deleted: $dihapus');

    // Server STATS
    final info = await client.stats();
    print('Available bytes: ${info['bytes']}');
    print('Current items: ${info['curr_items']}');
    print('Total hits: ${info['get_hits']}');
    print('Total misses: ${info['get_misses']}');

  } finally {
    await client.tutup();
  }
}

// Scenario: two processes try to update the same balance counter
Future<bool> updateSaldoAman(
  MemcachedClient client,
  String kunci,
  double delta,
) async {
  const maxRetry = 5;

  for (int percobaan = 0; percobaan < maxRetry; percobaan++) {
    // 1. Read the current value along with the CAS token
    final (:value, :casToken) = await client.gets(kunci);

    if (value == null || casToken == null) {
      // The key doesn't exist — create it
      final nilaiAwal = delta.toString();
      return await client.add(kunci, nilaiAwal);
    }

    // 2. Calculate the new value
    final saldoLama = double.tryParse(value) ?? 0;
    final saldoBaru = (saldoLama + delta).toStringAsFixed(2);

    // 3. Update with CAS — only succeeds if the token is still valid
    final berhasil = await client.cas(kunci, saldoBaru, casToken);

    if (berhasil) {
      print('Balance updated successfully: $saldoLama → $saldoBaru');
      return true;
    }

    // Invalid token — someone changed it first, try again
    print('CAS conflict, attempt ${percobaan + 1}/$maxRetry');
    await Future.delayed(Duration(milliseconds: 10 * (percobaan + 1)));
  }

  throw StateError('Failed to update after $maxRetry attempts');
}

import 'dart:convert';

class MemcachedCluster {
  final List<MemcachedClient> _nodes;
  final List<_VirtualNode> _ring = [];
  static const int _virtualNodesPerServer = 150;

  MemcachedCluster(List<String> servers)
      : _nodes = servers.map((s) {
          final parts = s.split(':');
          return MemcachedClient(
            host: parts[0],
            port: int.parse(parts[1]),
          );
        }).toList() {
    _bangunRing();
  }

  void _bangunRing() {
    for (int i = 0; i < _nodes.length; i++) {
      for (int v = 0; v < _virtualNodesPerServer; v++) {
        final kunci = 'node$i:virtual$v';
        final hash = _hash(kunci);
        _ring.add(_VirtualNode(hash: hash, nodeIndex: i));
      }
    }
    _ring.sort((a, b) => a.hash.compareTo(b.hash));
  }

  // Consistent hash — a simple MD5-like hash
  int _hash(String kunci) {
    int hash = 0;
    for (final char in kunci.codeUnits) {
      hash = (hash * 31 + char) & 0x7fffffff;
    }
    return hash;
  }

  // Determine the node based on the key (consistent hashing)
  MemcachedClient _pilihNode(String kunci) {
    if (_nodes.length == 1) return _nodes[0];

    final hash = _hash(kunci);

    // Find the first virtual node whose hash is >= the key hash
    for (final vNode in _ring) {
      if (vNode.hash >= hash) {
        return _nodes[vNode.nodeIndex];
      }
    }

    // Wrap-around: use the first node in the ring
    return _nodes[_ring.first.nodeIndex];
  }

  Future<void> inisialisasi() async {
    for (final node in _nodes) {
      await node.hubungkan();
    }
    print('Memcached cluster: ${_nodes.length} nodes connected');
  }

  // Operations automatically go to the right node
  Future<bool> set(String key, String value, {int exptime = 0}) {
    return _pilihNode(key).set(key, value, exptime: exptime);
  }

  Future<String?> get(String key) {
    return _pilihNode(key).get(key);
  }

  Future<bool> delete(String key) {
    return _pilihNode(key).delete(key);
  }

  Future<void> tutup() async {
    for (final node in _nodes) await node.tutup();
  }
}

class _VirtualNode {
  final int hash;
  final int nodeIndex;
  const _VirtualNode({required this.hash, required this.nodeIndex});
}

// Cluster usage
Future<void> main() async {
  final cluster = MemcachedCluster([
    'memcached-1:11211',
    'memcached-2:11211',
    'memcached-3:11211',
  ]);

  await cluster.inisialisasi();

  // Automatic distribution to nodes based on the key
  await cluster.set('user:1001', '{"nama": "Budi"}', exptime: 3600);
  await cluster.set('user:1002', '{"nama": "Siti"}', exptime: 3600);

  final user = await cluster.get('user:1001');
  print(user);

  await cluster.tutup();
}

Future<Map<String, dynamic>?> ambilProduk(
  MemcachedClient mc,
  String id,
  Future<Map<String, dynamic>?> Function(String) dariDB,
) async {
  final kunci = 'produk:$id';

  // 1. Check the cache
  final cached = await mc.get(kunci);
  if (cached != null) {
    return jsonDecode(cached) as Map<String, dynamic>;
  }

  // 2. Fetch from the database
  final produk = await dariDB(id);
  if (produk == null) {
    // Negative caching — prevent a thundering herd to the database
    // Store a "not found" marker for 60 seconds
    await mc.set(kunci, '__not_found__', exptime: 60);
    return null;
  }

  // 3. Store in the cache
  await mc.set(kunci, jsonEncode(produk), exptime: 3600);
  return produk;
}

// The thundering herd problem: many simultaneous requests hit the database
// when a cache entry expires

Future<Map<String, dynamic>?> ambilDenganLock(
  MemcachedClient mc,
  String id,
  Future<Map<String, dynamic>?> Function(String) dariDB,
) async {
  final kunci = 'produk:$id';
  final kunciLock = 'lock:produk:$id';

  // Check the cache
  final cached = await mc.get(kunci);
  if (cached != null && cached != '__loading__') {
    return jsonDecode(cached) as Map<String, dynamic>;
  }

  // Use ADD as a lock — only one succeeds
  final dapatLock = await mc.add(kunciLock, '1', exptime: 10);

  if (!dapatLock) {
    // Someone else is loading — wait a bit and try again
    await Future.delayed(Duration(milliseconds: 100));
    return ambilDenganLock(mc, id, dariDB); // retry
  }

  try {
    // We load the data
    final produk = await dariDB(id);
    if (produk != null) {
      await mc.set(kunci, jsonEncode(produk), exptime: 3600);
    }
    return produk;
  } finally {
    await mc.delete(kunciLock);
  }
}

// ANTI-PATTERN: Memcached rejects values > 1MB (default)
final dataBesar = List.generate(100000, (i) => {'id': i, 'data': 'xxx'});
await client.set('data_besar', jsonEncode(dataBesar));
// ✗ SERVER_ERROR object too large for cache

// CORRECT: split into chunks or use Redis, which is more flexible
final chunks = <String>[];
final json = jsonEncode(dataBesar);
for (int i = 0; i < json.length; i += 900000) {
  chunks.add(json.substring(i, (i + 900000).clamp(0, json.length)));
}

// Store the chunk count and each chunk
await client.set('data:chunks', chunks.length.toString());
for (int i = 0; i < chunks.length; i++) {
  await client.set('data:chunk:$i', chunks[i]);
}

// ANTI-PATTERN: keys > 250 characters aren't supported by Memcached
final keyPanjang = 'produk:kategori:elektronik:laptop:gaming:asus:rog:2024';
// Might still be okay, but...
final keyTerlalu = 'a' * 300; // ✗ error: key too long

// CORRECT: hash long keys into short keys
import 'dart:convert';
String hashKey(String panjang) {
  // MD5 or SHA256 for consistent, short keys
  final bytes = utf8.encode(panjang);
  // Use the crypto package for a proper hash
  return bytes.fold(0, (a, b) => a ^ b).toRadixString(16).padLeft(8, '0');
}

// ANTI-PATTERN: assuming the cache always has the value
final nilai = await client.get('produk:P001');
final harga = jsonDecode(nilai!)['harga']; // ✗ crashes on a cache miss (null)!

// CORRECT: always handle cache misses
final nilai = await client.get('produk:P001');
if (nilai == null) {
  // Cache miss — fetch from the database
  final produk = await database.ambilProduk('P001');
  if (produk != null) {
    await client.set('produk:P001', jsonEncode(produk), exptime: 3600);
  }
  return produk;
}
return jsonDecode(nilai) as Map<String, dynamic>;