hyper-social / hyper-feed / lib / feed-store.js

import { randomBytes } from 'crypto'

const SOURCES = ['irc', 'keet', 'mastodon', 'discord', 'email']
const TYPES = ['post', 'reply', 'boost', 'reaction', 'system']

class FeedStore {
  constructor (bee) {
    this.bee = bee
  }

  _genId () {
    return randomBytes(8).toString('hex')
  }

  _itemKey (id) {
    return `item/${id}`
  }

  _timelineKey (timestamp, id) {
    return `timeline/${this._padTimestamp(timestamp)}/${id}`
  }

  _sourceKey (source, timestamp, id) {
    return `source/${source}/${this._padTimestamp(timestamp)}/${id}`
  }

  _authorKey (authorId, timestamp, id) {
    return `author/${authorId}/${this._padTimestamp(timestamp)}/${id}`
  }

  _padTimestamp (ts) {
    return String(ts).padStart(16, '0')
  }

  async ready () {
    await this.bee.ready()
  }

  async add (opts = {}) {
    const id = opts.id || this._genId()
    const now = Date.now()

    const item = {
      id,
      source: opts.source || 'keet',
      author: opts.author || 'unknown',
      authorId: opts.authorId || opts.author || 'unknown',
      text: opts.text || '',
      timestamp: opts.timestamp || now,
      type: opts.type || 'post',
      tags: opts.tags || [],
      url: opts.url || null,
      metadata: opts.metadata || {}
    }

    const batch = this.bee.batch()

    // Primary record
    await batch.put(this._itemKey(id), item)

    // Timeline index (stores just the id for lookup)
    await batch.put(this._timelineKey(item.timestamp, id), { id })

    // Source index
    await batch.put(this._sourceKey(item.source, item.timestamp, id), { id })

    // Author index
    await batch.put(this._authorKey(item.authorId, item.timestamp, id), { id })

    await batch.flush()

    return item
  }

  async get (id) {
    const node = await this.bee.get(this._itemKey(id))
    return node ? node.value : null
  }

  async remove (id) {
    const item = await this.get(id)
    if (!item) return false

    const batch = this.bee.batch()

    // Remove primary record
    await batch.del(this._itemKey(id))

    // Remove timeline index
    await batch.del(this._timelineKey(item.timestamp, id))

    // Remove source index
    await batch.del(this._sourceKey(item.source, item.timestamp, id))

    // Remove author index
    await batch.del(this._authorKey(item.authorId, item.timestamp, id))

    await batch.flush()

    return true
  }

  async timeline (opts = {}) {
    const { limit, before, after, source, type } = opts
    const items = []

    const rangeOpts = {}

    if (after) {
      rangeOpts.gt = `timeline/${this._padTimestamp(after)}/~`
    } else {
      rangeOpts.gte = 'timeline/'
    }

    if (before) {
      rangeOpts.lt = `timeline/${this._padTimestamp(before)}/`
    } else {
      rangeOpts.lte = 'timeline/~'
    }

    for await (const node of this.bee.createReadStream(rangeOpts)) {
      if (!node.value) continue

      const item = await this.get(node.value.id)
      if (!item) continue

      // Filter by source
      if (source && item.source !== source) continue

      // Filter by type
      if (type && item.type !== type) continue

      items.push(item)

      if (limit && items.length >= limit) break
    }

    return items
  }

  async bySource (source, opts = {}) {
    const { limit, before } = opts
    const items = []

    const rangeOpts = {
      gte: `source/${source}/`
    }

    if (before) {
      rangeOpts.lt = `source/${source}/${this._padTimestamp(before)}/`
    } else {
      rangeOpts.lte = `source/${source}/~`
    }

    for await (const node of this.bee.createReadStream(rangeOpts)) {
      if (!node.value) continue

      const item = await this.get(node.value.id)
      if (!item) continue

      items.push(item)

      if (limit && items.length >= limit) break
    }

    return items
  }

  async byAuthor (authorId, opts = {}) {
    const { limit } = opts
    const items = []

    for await (const node of this.bee.createReadStream({
      gte: `author/${authorId}/`,
      lte: `author/${authorId}/~`
    })) {
      if (!node.value) continue

      const item = await this.get(node.value.id)
      if (!item) continue

      items.push(item)

      if (limit && items.length >= limit) break
    }

    return items
  }

  async search (query) {
    const q = query.toLowerCase()
    const results = []

    for await (const node of this.bee.createReadStream({
      gte: 'item/',
      lte: 'item/~'
    })) {
      const item = node.value
      if (!item) continue

      if (item.text && item.text.toLowerCase().includes(q)) {
        results.push(item)
        continue
      }

      if (item.tags && item.tags.some(t => t.toLowerCase().includes(q))) {
        results.push(item)
        continue
      }
    }

    return results
  }

  async stats () {
    const bySource = {}
    const byType = {}

    for await (const node of this.bee.createReadStream({
      gte: 'item/',
      lte: 'item/~'
    })) {
      const item = node.value
      if (!item) continue

      bySource[item.source] = (bySource[item.source] || 0) + 1
      byType[item.type] = (byType[item.type] || 0) + 1
    }

    let total = 0
    for (const count of Object.values(bySource)) {
      total += count
    }

    return {
      total,
      bySource,
      byType
    }
  }
}

export { FeedStore, SOURCES, TYPES }