redis_cache.nim 4.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156
  1. # SPDX-License-Identifier: AGPL-3.0-only
  2. import asyncdispatch, times, 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. # flatty can't serialize DateTime, so we need to define this
  13. proc toFlatty*(s: var string, x: DateTime) =
  14. s.toFlatty(x.toTime().toUnix())
  15. proc fromFlatty*(s: string, i: var int, x: var DateTime) =
  16. var unix: int64
  17. s.fromFlatty(i, unix)
  18. x = fromUnix(unix).utc()
  19. proc setCacheTimes*(cfg: Config) =
  20. rssCacheTime = cfg.rssCacheTime * 60
  21. listCacheTime = cfg.listCacheTime * 60
  22. proc migrate*(key, match: string) {.async.} =
  23. pool.withAcquire(r):
  24. let hasKey = await r.get(key)
  25. if hasKey == redisNil:
  26. let list = await r.scan(newCursor(0), match, 100000)
  27. r.startPipelining()
  28. for item in list:
  29. discard await r.del(item)
  30. await r.setk(key, "true")
  31. discard await r.flushPipeline()
  32. proc initRedisPool*(cfg: Config) {.async.} =
  33. try:
  34. pool = await newRedisPool(cfg.redisConns, cfg.redisMaxConns,
  35. host=cfg.redisHost, port=cfg.redisPort,
  36. password=cfg.redisPassword)
  37. await migrate("flatty", "*:*")
  38. await migrate("snappyRss", "rss:*")
  39. await migrate("userBuckets", "p:*")
  40. await migrate("profileDates", "p:*")
  41. await migrate("profileStats", "p:*")
  42. pool.withAcquire(r):
  43. # optimize memory usage for profile ID buckets
  44. await r.configSet("hash-max-ziplist-entries", "1000")
  45. except OSError:
  46. stdout.write "Failed to connect to Redis.\n"
  47. stdout.flushFile
  48. quit(1)
  49. template pidKey(name: string): string = "pid:" & $(hash(name) div 1_000_000)
  50. template profileKey(name: string): string = "p:" & name
  51. template listKey(l: List): string = "l:" & l.id
  52. proc get(query: string): Future[string] {.async.} =
  53. pool.withAcquire(r):
  54. result = await r.get(query)
  55. proc setEx(key: string; time: int; data: string) {.async.} =
  56. pool.withAcquire(r):
  57. discard await r.setEx(key, time, data)
  58. proc cache*(data: List) {.async.} =
  59. await setEx(data.listKey, listCacheTime, compress(toFlatty(data)))
  60. proc cache*(data: PhotoRail; name: string) {.async.} =
  61. await setEx("pr:" & toLower(name), baseCacheTime, compress(toFlatty(data)))
  62. proc cache*(data: Profile) {.async.} =
  63. if data.username.len == 0 or data.id.len == 0: return
  64. let name = toLower(data.username)
  65. pool.withAcquire(r):
  66. r.startPipelining()
  67. discard await r.setEx(name.profileKey, baseCacheTime, compress(toFlatty(data)))
  68. discard await r.setEx("i:" & data.id , baseCacheTime, data.username)
  69. discard await r.hSet(name.pidKey, name, data.id)
  70. discard await r.flushPipeline()
  71. proc cacheProfileId*(username, id: string) {.async.} =
  72. if username.len == 0 or id.len == 0: return
  73. let name = toLower(username)
  74. pool.withAcquire(r):
  75. discard await r.hSet(name.pidKey, name, id)
  76. proc cacheRss*(query: string; rss: Rss) {.async.} =
  77. let key = "rss:" & query
  78. pool.withAcquire(r):
  79. r.startPipelining()
  80. discard await r.hSet(key, "rss", rss.feed)
  81. discard await r.hSet(key, "min", rss.cursor)
  82. discard await r.expire(key, rssCacheTime)
  83. discard await r.flushPipeline()
  84. proc getProfileId*(username: string): Future[string] {.async.} =
  85. let name = toLower(username)
  86. pool.withAcquire(r):
  87. result = await r.hGet(name.pidKey, name)
  88. if result == redisNil:
  89. result.setLen(0)
  90. proc getCachedProfile*(username: string; fetch=true): Future[Profile] {.async.} =
  91. let prof = await get("p:" & toLower(username))
  92. if prof != redisNil:
  93. result = fromFlatty(uncompress(prof), Profile)
  94. elif fetch:
  95. result = await getProfile(username)
  96. proc getCachedProfileUsername*(userId: string): Future[string] {.async.} =
  97. let username = await get("i:" & userId)
  98. if username != redisNil:
  99. result = username
  100. else:
  101. let profile = await getProfileById(userId)
  102. result = profile.username
  103. await cache(profile)
  104. proc getCachedPhotoRail*(name: string): Future[PhotoRail] {.async.} =
  105. if name.len == 0: return
  106. let rail = await get("pr:" & toLower(name))
  107. if rail != redisNil:
  108. result = fromFlatty(uncompress(rail), PhotoRail)
  109. else:
  110. result = await getPhotoRail(name)
  111. await cache(result, name)
  112. proc getCachedList*(username=""; slug=""; id=""): Future[List] {.async.} =
  113. let list = if id.len == 0: redisNil
  114. else: await get("l:" & id)
  115. if list != redisNil:
  116. result = fromFlatty(uncompress(list), List)
  117. else:
  118. if id.len > 0:
  119. result = await getGraphList(id)
  120. else:
  121. result = await getGraphListBySlug(username, slug)
  122. await cache(result)
  123. proc getCachedRss*(key: string): Future[Rss] {.async.} =
  124. let k = "rss:" & key
  125. pool.withAcquire(r):
  126. result.cursor = await r.hGet(k, "min")
  127. if result.cursor.len > 2:
  128. result.feed = await r.hGet(k, "rss")
  129. else:
  130. result.cursor.setLen 0