redis_cache.nim 5.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187
  1. # SPDX-License-Identifier: AGPL-3.0-only
  2. import asyncdispatch, times, strformat, strutils, tables, hashes
  3. import redis, redpool, flatty, supersnappy
  4. import types, api
  5. const
  6. redisNil = "\0\0"
  7. baseCacheTime = 60 * 60
  8. var
  9. pool: RedisPool
  10. rssCacheTime: int
  11. listCacheTime*: int
  12. template dawait(future) =
  13. discard await future
  14. # flatty can't serialize DateTime, so we need to define this
  15. proc toFlatty*(s: var string, x: DateTime) =
  16. s.toFlatty(x.toTime().toUnix())
  17. proc fromFlatty*(s: string, i: var int, x: var DateTime) =
  18. var unix: int64
  19. s.fromFlatty(i, unix)
  20. x = fromUnix(unix).utc()
  21. proc setCacheTimes*(cfg: Config) =
  22. rssCacheTime = cfg.rssCacheTime * 60
  23. listCacheTime = cfg.listCacheTime * 60
  24. proc migrate*(key, match: string) {.async.} =
  25. pool.withAcquire(r):
  26. let hasKey = await r.get(key)
  27. if hasKey == redisNil:
  28. let list = await r.scan(newCursor(0), match, 100000)
  29. r.startPipelining()
  30. for item in list:
  31. dawait r.del(item)
  32. await r.setk(key, "true")
  33. dawait r.flushPipeline()
  34. proc initRedisPool*(cfg: Config) {.async.} =
  35. try:
  36. pool = await newRedisPool(cfg.redisConns, cfg.redisMaxConns,
  37. host=cfg.redisHost, port=cfg.redisPort,
  38. password=cfg.redisPassword)
  39. await migrate("flatty", "*:*")
  40. await migrate("snappyRss", "rss:*")
  41. await migrate("userBuckets", "p:*")
  42. await migrate("profileDates", "p:*")
  43. await migrate("profileStats", "p:*")
  44. await migrate("userType", "p:*")
  45. pool.withAcquire(r):
  46. # optimize memory usage for user ID buckets
  47. await r.configSet("hash-max-ziplist-entries", "1000")
  48. except OSError:
  49. stdout.write "Failed to connect to Redis.\n"
  50. stdout.flushFile
  51. quit(1)
  52. template uidKey(name: string): string = "pid:" & $(hash(name) div 1_000_000)
  53. template userKey(name: string): string = "p:" & name
  54. template listKey(l: List): string = "l:" & l.id
  55. template tweetKey(id: int64): string = "t:" & $id
  56. proc get(query: string): Future[string] {.async.} =
  57. pool.withAcquire(r):
  58. result = await r.get(query)
  59. proc setEx(key: string; time: int; data: string) {.async.} =
  60. pool.withAcquire(r):
  61. dawait r.setEx(key, time, data)
  62. proc cacheUserId(username, id: string) {.async.} =
  63. if username.len == 0 or id.len == 0: return
  64. let name = toLower(username)
  65. pool.withAcquire(r):
  66. dawait r.hSet(name.uidKey, name, id)
  67. proc cache*(data: List) {.async.} =
  68. await setEx(data.listKey, listCacheTime, compress(toFlatty(data)))
  69. proc cache*(data: PhotoRail; name: string) {.async.} =
  70. await setEx("pr:" & toLower(name), baseCacheTime, compress(toFlatty(data)))
  71. proc cache*(data: User) {.async.} =
  72. if data.username.len == 0: return
  73. let name = toLower(data.username)
  74. await cacheUserId(name, data.id)
  75. pool.withAcquire(r):
  76. dawait r.setEx(name.userKey, baseCacheTime, compress(toFlatty(data)))
  77. proc cache*(data: Tweet) {.async.} =
  78. if data.isNil or data.id == 0: return
  79. pool.withAcquire(r):
  80. dawait r.setEx(data.id.tweetKey, baseCacheTime, compress(toFlatty(data)))
  81. proc cacheRss*(query: string; rss: Rss) {.async.} =
  82. let key = "rss:" & query
  83. pool.withAcquire(r):
  84. dawait r.hSet(key, "rss", rss.feed)
  85. dawait r.hSet(key, "min", rss.cursor)
  86. dawait r.expire(key, rssCacheTime)
  87. template deserialize(data, T) =
  88. try:
  89. result = fromFlatty(uncompress(data), T)
  90. except:
  91. echo "Decompression failed($#): '$#'" % [astToStr(T), data]
  92. proc getUserId*(username: string): Future[string] {.async.} =
  93. let name = toLower(username)
  94. pool.withAcquire(r):
  95. result = await r.hGet(name.uidKey, name)
  96. if result == redisNil:
  97. let user = await getUser(username)
  98. if user.suspended:
  99. return "suspended"
  100. else:
  101. await cacheUserId(name, user.id)
  102. return user.id
  103. proc getCachedUser*(username: string; fetch=true): Future[User] {.async.} =
  104. let prof = await get("p:" & toLower(username))
  105. if prof != redisNil:
  106. prof.deserialize(User)
  107. elif fetch:
  108. let userId = await getUserId(username)
  109. result = await getGraphUser(userId)
  110. await cache(result)
  111. proc getCachedUsername*(userId: string): Future[string] {.async.} =
  112. let
  113. key = "i:" & userId
  114. username = await get(key)
  115. if username != redisNil:
  116. result = username
  117. else:
  118. let user = await getUserById(userId)
  119. result = user.username
  120. await setEx(key, baseCacheTime, result)
  121. proc getCachedTweet*(id: int64): Future[Tweet] {.async.} =
  122. if id == 0: return
  123. let tweet = await get(id.tweetKey)
  124. if tweet != redisNil:
  125. tweet.deserialize(Tweet)
  126. else:
  127. result = await getStatus($id)
  128. if result.isNil:
  129. await cache(result)
  130. proc getCachedPhotoRail*(name: string): Future[PhotoRail] {.async.} =
  131. if name.len == 0: return
  132. let rail = await get("pr:" & toLower(name))
  133. if rail != redisNil:
  134. rail.deserialize(PhotoRail)
  135. else:
  136. result = await getPhotoRail(name)
  137. await cache(result, name)
  138. proc getCachedList*(username=""; slug=""; id=""): Future[List] {.async.} =
  139. let list = if id.len == 0: redisNil
  140. else: await get("l:" & id)
  141. if list != redisNil:
  142. list.deserialize(List)
  143. else:
  144. if id.len > 0:
  145. result = await getGraphList(id)
  146. else:
  147. result = await getGraphListBySlug(username, slug)
  148. await cache(result)
  149. proc getCachedRss*(key: string): Future[Rss] {.async.} =
  150. let k = "rss:" & key
  151. pool.withAcquire(r):
  152. result.cursor = await r.hGet(k, "min")
  153. if result.cursor.len > 2:
  154. result.feed = await r.hGet(k, "rss")
  155. else:
  156. result.cursor.setLen 0