redis_cache.nim 4.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132
  1. import asyncdispatch, times, strutils, tables, hashes
  2. import redis, redpool, frosty, supersnappy
  3. import types, api
  4. const redisNil = "\0\0"
  5. var
  6. pool {.threadvar.}: RedisPool
  7. baseCacheTime = 60 * 60
  8. rssCacheTime: int
  9. listCacheTime*: int
  10. proc setCacheTimes*(cfg: Config) =
  11. rssCacheTime = cfg.rssCacheTime * 60
  12. listCacheTime = cfg.listCacheTime * 60
  13. proc migrate*(key, match: string) {.async.} =
  14. pool.withAcquire(r):
  15. let hasKey = await r.get(key)
  16. if hasKey == redisNil:
  17. let list = await r.scan(newCursor(0), match, 100000)
  18. r.startPipelining()
  19. for item in list:
  20. if item != "p:" or item == match:
  21. discard await r.del(item)
  22. await r.setk(key, "true")
  23. discard await r.flushPipeline()
  24. proc initRedisPool*(cfg: Config) {.async.} =
  25. try:
  26. pool = await newRedisPool(cfg.redisConns, maxConns=cfg.redisMaxConns,
  27. host=cfg.redisHost, port=cfg.redisPort, password=cfg.redisPassword)
  28. await migrate("snappyRss", "rss:*")
  29. await migrate("oldFrosty", "*")
  30. await migrate("userBuckets", "p:")
  31. pool.withAcquire(r):
  32. await r.configSet("hash-max-ziplist-entries", "1000")
  33. except OSError:
  34. stdout.write "Failed to connect to Redis.\n"
  35. stdout.flushFile
  36. quit(1)
  37. template pidKey(name: string): string = "pid:" & $(hash(name) div 1_000_000)
  38. template profileKey(name: string): string = "p:" & name
  39. template listKey(l: List): string = toLower("l:" & l.username & '/' & l.name)
  40. proc get(query: string): Future[string] {.async.} =
  41. pool.withAcquire(r):
  42. result = await r.get(query)
  43. proc setex(key: string; time: int; data: string) {.async.} =
  44. pool.withAcquire(r):
  45. discard await r.setex(key, time, data)
  46. proc cache*(data: List) {.async.} =
  47. await setex(data.listKey, listCacheTime, compress(freeze(data)))
  48. proc cache*(data: PhotoRail; name: string) {.async.} =
  49. await setex("pr:" & name, baseCacheTime, compress(freeze(data)))
  50. proc cache*(data: Profile) {.async.} =
  51. if data.username.len == 0 or data.id.len == 0: return
  52. let name = toLower(data.username)
  53. pool.withAcquire(r):
  54. r.startPipelining()
  55. discard await r.setex(name.profileKey, baseCacheTime, compress(freeze(data)))
  56. discard await r.hset(name.pidKey, name, data.id)
  57. discard await r.flushPipeline()
  58. proc cacheProfileId*(username, id: string) {.async.} =
  59. if username.len == 0 or id.len == 0: return
  60. let name = toLower(username)
  61. pool.withAcquire(r):
  62. discard await r.hset(name.pidKey, name, id)
  63. proc cacheRss*(query: string; rss: Rss) {.async.} =
  64. let key = "rss:" & query
  65. pool.withAcquire(r):
  66. r.startPipelining()
  67. discard await r.hset(key, "rss", rss.feed)
  68. discard await r.hset(key, "min", rss.cursor)
  69. discard await r.expire(key, rssCacheTime)
  70. discard await r.flushPipeline()
  71. proc getProfileId*(username: string): Future[string] {.async.} =
  72. let name = toLower(username)
  73. pool.withAcquire(r):
  74. result = await r.hget(name.pidKey, name)
  75. if result == redisNil:
  76. result.setLen(0)
  77. proc getCachedProfile*(username: string; fetch=true): Future[Profile] {.async.} =
  78. let prof = await get("p:" & toLower(username))
  79. if prof != redisNil:
  80. uncompress(prof).thaw(result)
  81. elif fetch:
  82. result = await getProfile(username)
  83. proc getCachedPhotoRail*(name: string): Future[PhotoRail] {.async.} =
  84. if name.len == 0: return
  85. let rail = await get("pr:" & toLower(name))
  86. if rail != redisNil:
  87. uncompress(rail).thaw(result)
  88. else:
  89. result = await getPhotoRail(name)
  90. await cache(result, name)
  91. proc getCachedList*(username=""; name=""; id=""): Future[List] {.async.} =
  92. let list = if id.len > 0: redisNil
  93. else: await get(toLower("l:" & username & '/' & name))
  94. if list != redisNil:
  95. uncompress(list).thaw(result)
  96. else:
  97. if id.len > 0:
  98. result = await getGraphListById(id)
  99. else:
  100. result = await getGraphList(username, name)
  101. await cache(result)
  102. proc getCachedRss*(key: string): Future[Rss] {.async.} =
  103. let k = "rss:" & key
  104. pool.withAcquire(r):
  105. result.cursor = await r.hget(k, "min")
  106. if result.cursor.len > 2:
  107. result.feed = await r.hget(k, "rss")
  108. else:
  109. result.cursor.setLen 0