macltok

use std::{collections::HashMap, path::PathBuf, sync::Arc};

use sha2::{Digest, Sha256};

use tantivy::{

    DocAddress, DocId, DocSet, Index, IndexSettings, ReloadPolicy, Score, Searcher, SegmentReader,

    TantivyDocument, Term,

    collector::{Collector, SegmentCollector},

    directory::{Directory, OwnedBytes, RamDirectory},

    fastfield::{AliveBitSet, write_alive_bitset},

    indexer::{SingleSegmentIndexWriter, merge_filtered_segments},

    query::{BooleanQuery, Occur, Query, TermSetQuery},

    schema::{

        FAST, Field, IndexRecordOption, STORED, STRING, Schema, TextFieldIndexing, TextOptions,

        Value,

    },

};

use tantivy_common::BitSet;

use super::{

    MessageIndex,

    text::{TOKENIZER, analyzer, build_query},

};

use crate::{Error, ErrorKind};

const WRITER_BUDGET: usize = 4_000_000;

/// segments of a size class merge once there are this many of them

const MERGE_FACTOR: usize = 6;

/// decompressed docstore blocks a searcher keeps per segment, tantivy's default is 100

const DOCSTORE_BLOCKS: usize = 2;

#[derive(Clone, Debug, Eq, PartialEq)]

pub(crate) struct Doc {

    pub(crate) key: String,

    pub(crate) room: String,

    pub(crate) sender: String,

    /// the original's, or the edit's while the original hasn't turned up

    pub(crate) ts: u64,

    /// the event whose text is indexed, the original or its latest edit

    pub(crate) content: String,

    pub(crate) content_ts: u64,

    /// only kept for provisional docs, read back empty otherwise

    pub(crate) text: String,

    /// an edit seen before its original, under `provisional_key`. it's kept out of results,

    /// since its timestamp moves to the original's once that turns up

    pub(crate) provisional: bool,

}

/// one per target and edit sender, so an edit from anyone else can't crowd out the author's

pub(crate) fn provisional_key(target: &str, sender: &str) -> String {

    format!("{target}\n{sender}")

}

#[derive(Clone, Debug, Eq, PartialEq)]

pub(crate) struct Hit {

    pub(crate) key: String,

    pub(crate) room: String,

    pub(crate) sender: String,

    pub(crate) content: String,

    pub(crate) ts: u64,

    pub(crate) tie: u64,

}

struct SegmentRecord {

    id: u64,

    del_gen: u64,

}

#[derive(Default)]

pub(crate) struct Changes {

    pub(crate) puts: Vec<(String, Vec<u8>)>,

    pub(crate) deletes: Vec<String>,

    /// the index as it stands once these land, which is what has to be saved with them

    pub(crate) manifest: serde_json::Value,

    files: Vec<u64>,

    dels: Vec<(u64, u64)>,

    removed: Vec<u64>,

}

#[derive(Clone, Copy)]

struct Fields {

    key: Field,

    room: Field,

    sender: Field,

    content: Field,

    text: Field,

    ts: Field,

    tie: Field,

    content_ts: Field,

    pending: Field,

    target: Field,

}

struct Segment {

    id: u64,

    raw: RamDirectory,

    index: Index,

    searcher: Searcher,

    alive: Arc<BitSet>,

    max_doc: u32,

    del_gen: u64,

    persisted: bool,

    deletes_dirty: bool,

}

impl Segment {

    fn live(&self) -> usize {

        self.alive.len()

    }

}

pub(crate) struct KeywordIndex {

    schema: Schema,

    fields: Fields,

    settings: IndexSettings,

    segments: Vec<Segment>,

    buffer: HashMap<String, Doc>,

    next_id: u64,

    persistent: bool,

    /// merged away since the last commit, with the delete generation they were persisted at

    removed: Vec<(u64, bool, u64)>,

}

fn schema() -> (Schema, Fields) {

    let mut builder = Schema::builder();

    let text_options = TextOptions::default().set_indexing_options(

        TextFieldIndexing::default()

            .set_tokenizer(TOKENIZER)

            .set_index_option(IndexRecordOption::WithFreqsAndPositions),

    );

    let fields = Fields {

        key: builder.add_text_field("key", STRING | STORED),

        room: builder.add_text_field("room", STRING | STORED),

        sender: builder.add_text_field("sender", STRING | STORED),

        content: builder.add_text_field("content", STRING | STORED),

        text: builder.add_text_field("text", text_options),

        ts: builder.add_u64_field("ts", FAST | STORED),

        tie: builder.add_u64_field("tie", FAST),

        content_ts: builder.add_u64_field("content_ts", STORED),

        pending: builder.add_text_field("pending", STORED),

        target: builder.add_text_field("target", STRING),

    };

    (builder.build(), fields)

}

fn tie(key: &str) -> u64 {

    let digest = Sha256::digest(key.as_bytes());

    u64::from_le_bytes(digest[..8].try_into().expect("a sha256 digest is 32 bytes"))

}

fn settings() -> IndexSettings {

    // the dedicated compressor is a thread, and wasm32-unknown-unknown has none

    IndexSettings {

        docstore_compress_dedicated_thread: false,

        ..IndexSettings::default()

    }

}

fn tantivy_error(

    context: &'static str,

    error: impl std::error::Error + Send + Sync + 'static,

) -> Error {

    Error::with_source(ErrorKind::Store, context, error)

}

fn open_segment(id: u64, raw: RamDirectory, max_doc_hint: Option<u32>) -> Result<Segment, Error> {

    let index = Index::open(raw.clone())

        .map_err(|error| tantivy_error("opening a search segment", error))?;

    index.tokenizers().register(TOKENIZER, analyzer());

    let reader = index

        .reader_builder()

        .reload_policy(ReloadPolicy::Manual)

        .doc_store_cache_num_blocks(DOCSTORE_BLOCKS)

        .try_into()

        .map_err(|error| tantivy_error("reading a search segment", error))?;

    let searcher = reader.searcher();

    let max_doc = searcher

        .segment_readers()

        .iter()

        .map(SegmentReader::max_doc)

        .sum();

    if searcher.segment_readers().len() > 1 || max_doc_hint.is_some_and(|hint| hint != max_doc) {

        return Err(Error::new(

            ErrorKind::Store,

            "a search segment has an unexpected shape",

        ));

    }

    Ok(Segment {

        id,

        raw,

        index,

        searcher,

        alive: Arc::new(BitSet::with_max_value_and_full(max_doc)),

        max_doc,

        del_gen: 0,

        persisted: false,

        deletes_dirty: false,

    })

}

fn pack(files: &[(String, OwnedBytes)]) -> Vec<u8> {

    let mut packed = Vec::with_capacity(

        files

            .iter()

            .map(|(name, bytes)| name.len() + bytes.len() + 12)

            .sum::<usize>()

            + 68,

    );

    packed.extend_from_slice(&(files.len() as u32).to_le_bytes());

    for (name, bytes) in files {

        packed.extend_from_slice(&(name.len() as u32).to_le_bytes());

        packed.extend_from_slice(name.as_bytes());

        packed.extend_from_slice(&(bytes.len() as u64).to_le_bytes());

        packed.extend_from_slice(bytes.as_slice());

    }

    packed

}

fn unpack(mut bytes: &[u8]) -> Option<Vec<(String, &[u8])>> {

    fn take<'a>(bytes: &mut &'a [u8], len: usize) -> Option<&'a [u8]> {

        if bytes.len() < len {

            return None;

        }

        let (head, tail) = bytes.split_at(len);

        *bytes = tail;

        Some(head)

    }

    let count = u32::from_le_bytes(take(&mut bytes, 4)?.try_into().ok()?);

    let mut files = Vec::new();

    for _ in 0..count {

        let name_len = u32::from_le_bytes(take(&mut bytes, 4)?.try_into().ok()?) as usize;

        let name = std::str::from_utf8(take(&mut bytes, name_len)?)

            .ok()?

            .to_owned();

        let len =

            usize::try_from(u64::from_le_bytes(take(&mut bytes, 8)?.try_into().ok()?)).ok()?;

        files.push((name, take(&mut bytes, len)?));

    }

    bytes.is_empty().then_some(files)

}

pub(crate) fn segment_key(id: u64) -> String {

    format!("seg/{id}")

}

pub(crate) fn deletes_key(id: u64, generation: u64) -> String {

    format!("del/{id}/{generation}")

}

fn manifest(records: impl Iterator<Item = SegmentRecord>, next_id: u64) -> serde_json::Value {

    let segments: Vec<_> = records

        .map(|record| serde_json::json!([record.id, record.del_gen]))

        .collect();

    serde_json::json!({ "segments": segments, "next": next_id })

}

impl KeywordIndex {

    pub(crate) fn new(persistent: bool) -> Self {

        let (schema, fields) = schema();

        Self {

            schema,

            fields,

            settings: settings(),

            segments: Vec::new(),

            buffer: HashMap::new(),

            next_id: 1,

            persistent,

            removed: Vec::new(),

        }

    }

    /// `blob` hands over what an earlier commit put, once each, and any of it missing or

    /// unreadable fails the load

    pub(crate) fn load(

        manifest: &serde_json::Value,

        mut blob: impl FnMut(&str) -> Option<Vec<u8>>,

    ) -> Result<Self, Error> {

        let invalid = || Error::new(ErrorKind::Store, "the search manifest is unreadable");

        let records = manifest["segments"]

            .as_array()

            .ok_or_else(invalid)?

            .iter()

            .map(|record| {

                Some(SegmentRecord {

                    id: record.get(0)?.as_u64()?,

                    del_gen: record.get(1)?.as_u64()?,

                })

            })

            .collect::<Option<Vec<_>>>()

            .ok_or_else(invalid)?;

        let next_id = manifest["next"].as_u64().ok_or_else(invalid)?;

        let mut index = Self::new(true);

        let missing = || {

            Error::new(

                ErrorKind::Store,

                "a search segment is missing or unreadable",

            )

        };

        for record in &records {

            let packed = blob(&segment_key(record.id)).ok_or_else(missing)?;

            let raw = RamDirectory::create();

            for (name, bytes) in unpack(&packed).ok_or_else(missing)? {

                raw.atomic_write(&PathBuf::from(name), bytes)

                    .map_err(|error| tantivy_error("loading a search segment", error))?;

            }

            drop(packed);

            let mut segment = open_segment(record.id, raw, None)?;

            if segment.index.schema() != index.schema {

                return Err(Error::new(

                    ErrorKind::Store,

                    "a search segment has another schema",

                ));

            }

            if record.del_gen > 0 {

                let bytes = blob(&deletes_key(record.id, record.del_gen)).ok_or_else(missing)?;

                let alive = AliveBitSet::open(OwnedBytes::new(bytes));

                let mut bits = BitSet::with_max_value(segment.max_doc);

                for doc in alive.iter_alive().filter(|doc| *doc < segment.max_doc) {

                    bits.insert(doc);

                }

                segment.alive = Arc::new(bits);

            }

            segment.del_gen = record.del_gen;

            segment.persisted = true;

            index.segments.push(segment);

        }

        index.next_id = next_id.max(

            records

                .iter()

                .map(|record| record.id + 1)

                .max()

                .unwrap_or(1),

        );

        Ok(index)

    }

    /// the blob names what is persisted points at

    #[cfg(test)]

    pub(crate) fn referenced(&self) -> Vec<String> {

        let mut keys = Vec::new();

        for segment in self.segments.iter().filter(|segment| segment.persisted) {

            keys.push(segment_key(segment.id));

            if segment.del_gen > 0 {

                keys.push(deletes_key(segment.id, segment.del_gen));

            }

        }

        keys

    }

    fn document(&self, doc: &Doc) -> TantivyDocument {

        let fields = self.fields;

        let mut document = TantivyDocument::default();

        document.add_text(fields.key, &doc.key);

        document.add_text(fields.room, &doc.room);

        document.add_text(fields.sender, &doc.sender);

        document.add_text(fields.content, &doc.content);

        document.add_u64(fields.ts, doc.ts);

        document.add_u64(fields.tie, tie(&doc.key));

        document.add_u64(fields.content_ts, doc.content_ts);

        if doc.provisional {

            document.add_text(fields.pending, &doc.text);

            let target = doc

                .key

                .split_once('\n')

                .map_or(doc.key.as_str(), |(target, _)| target);

            document.add_text(fields.target, target);

        } else {

            document.add_text(fields.text, &doc.text);

        }

        document

    }

    fn stored(&self, document: &TantivyDocument) -> Option<Doc> {

        let fields = self.fields;

        let text = |field| {

            document

                .get_first(field)

                .and_then(|value| value.as_str())

                .map(str::to_owned)

        };

        let number = |field| document.get_first(field).and_then(|value| value.as_u64());

        let pending = text(fields.pending);

        Some(Doc {

            key: text(fields.key)?,

            room: text(fields.room)?,

            sender: text(fields.sender)?,

            ts: number(fields.ts)?,

            content: text(fields.content)?,

            content_ts: number(fields.content_ts)?,

            provisional: pending.is_some(),

            text: pending.unwrap_or_default(),

        })

    }

    /// the live docs in `segment` whose `field` holds `value`

    fn find(&self, segment: &Segment, field: Field, value: &str) -> Vec<DocId> {

        let term = Term::from_field_text(field, value);

        let mut found = Vec::new();

        for reader in segment.searcher.segment_readers() {

            let Ok(inverted) = reader.inverted_index(field) else {

                continue;

            };

            let Ok(Some(mut postings)) = inverted.read_postings(&term, IndexRecordOption::Basic)

            else {

                continue;

            };

            let mut doc = postings.doc();

            while doc != tantivy::TERMINATED {

                if segment.alive.contains(doc) {

                    found.push(doc);

                }

                doc = postings.advance();

            }

        }

        found

    }

    fn lookup(&self, field: Field, value: &str) -> Option<Doc> {

        self.segments.iter().rev().find_map(|segment| {

            let doc = *self.find(segment, field, value).first()?;

            let document: TantivyDocument = segment.searcher.doc(DocAddress::new(0, doc)).ok()?;

            self.stored(&document)

        })

    }

    /// tombstones every live doc in a segment whose `field` holds `value`

    fn remove_where(&mut self, field: Field, value: &str) -> bool {

        let mut removed = false;

        for position in 0..self.segments.len() {

            let docs = self.find(&self.segments[position], field, value);

            if docs.is_empty() {

                continue;

            }

            let segment = &mut self.segments[position];

            let alive = Arc::make_mut(&mut segment.alive);

            for doc in docs {

                alive.remove(doc);

            }

            segment.deletes_dirty = true;

            removed = true;

        }

        removed

    }

    fn flush_buffer(&mut self) -> Result<(), Error> {

        if self.buffer.is_empty() {

            return Ok(());

        }

        let raw = RamDirectory::create();

        let index = Index::create(raw.clone(), self.schema.clone(), self.settings.clone())

            .map_err(|error| tantivy_error("creating a search segment", error))?;

        index.tokenizers().register(TOKENIZER, analyzer());

        let mut writer = SingleSegmentIndexWriter::new(index, WRITER_BUDGET)

            .map_err(|error| tantivy_error("writing a search segment", error))?;

        let mut docs: Vec<_> = self.buffer.values().collect();

        docs.sort_by_key(|doc| doc.ts);

        for doc in docs {

            writer

                .add_document(self.document(doc))

                .map_err(|error| tantivy_error("writing a search segment", error))?;

        }

        writer

            .finalize()

            .map_err(|error| tantivy_error("finishing a search segment", error))?;

        let segment = open_segment(self.next_id, raw, Some(self.buffer.len() as u32))?;

        self.next_id += 1;

        self.segments.push(segment);

        self.buffer.clear();

        Ok(())

    }

    fn class(live: usize) -> u32 {

        live.max(1).ilog(MERGE_FACTOR)

    }

    fn forget(&mut self, segment: &Segment) {

        if self.persistent {

            self.removed

                .push((segment.id, segment.persisted, segment.del_gen));

        }

    }

    /// merges a size class once it fills up, and rewrites a segment that's mostly deleted

    fn compact(&mut self) -> Result<(), Error> {

        let (empty, kept): (Vec<_>, Vec<_>) = std::mem::take(&mut self.segments)

            .into_iter()

            .partition(|segment| segment.live() == 0);

        self.segments = kept;

        for segment in &empty {

            self.forget(segment);

        }

        loop {

            let mut classes: HashMap<u32, Vec<usize>> = HashMap::new();

            for (position, segment) in self.segments.iter().enumerate() {

                classes

                    .entry(Self::class(segment.live()))

                    .or_default()

                    .push(position);

            }

            let full = classes

                .into_values()

                .find(|positions| positions.len() >= MERGE_FACTOR);

            let sparse = self

                .segments

                .iter()

                .position(|segment| {

                    segment.max_doc >= 1_000 && segment.live() * 2 < segment.max_doc as usize

                })

                .map(|position| vec![position]);

            let Some(positions) = full.or(sparse) else {

                return Ok(());

            };

            self.merge(&positions)?;

        }

    }

    fn merge(&mut self, positions: &[usize]) -> Result<(), Error> {

        let mut inputs = Vec::new();

        let mut alive = Vec::new();

        for position in positions {

            let segment = &self.segments[*position];

            let searchable = segment

                .index

                .searchable_segments()

                .map_err(|error| tantivy_error("merging search segments", error))?;

            let mut bytes = Vec::new();

            write_alive_bitset(&segment.alive, &mut bytes)

                .map_err(|error| tantivy_error("merging search segments", error))?;

            let bytes = OwnedBytes::new(bytes);

            for tantivy_segment in searchable {

                inputs.push(tantivy_segment);

                alive.push(Some(AliveBitSet::open(bytes.clone())));

            }

        }

        let raw = RamDirectory::create();

        merge_filtered_segments(&inputs, self.settings.clone(), alive, raw.clone())

            .map_err(|error| tantivy_error("merging search segments", error))?;

        drop(inputs);

        let merged = open_segment(self.next_id, raw, None)?;

        self.next_id += 1;

        let (merging, kept): (Vec<_>, Vec<_>) = std::mem::take(&mut self.segments)

            .into_iter()

            .enumerate()

            .partition(|(position, _)| positions.contains(position));

        self.segments = kept.into_iter().map(|(_, segment)| segment).collect();

        for (_, segment) in &merging {

            self.forget(segment);

        }

        self.segments.push(merged);

        Ok(())

    }

    fn segment_files(segment: &Segment) -> Result<Vec<u8>, Error> {

        let mut names: Vec<PathBuf> = segment

            .index

            .searchable_segment_metas()

            .map_err(|error| tantivy_error("reading a search segment", error))?

            .iter()

            .flat_map(|meta| meta.list_files())

            .collect();

        names.push(PathBuf::from("meta.json"));

        let mut files = Vec::new();

        for name in names {

            if !segment.raw.exists(&name).unwrap_or(false) {

                continue;

            }

            let bytes = segment

                .raw

                .open_read(&name)

                .map_err(|error| tantivy_error("reading a search segment", error))?

                .read_bytes()

                .map_err(|error| tantivy_error("reading a search segment", error))?;

            files.push((name.to_string_lossy().into_owned(), bytes));

        }

        Ok(pack(&files))

    }

}

impl MessageIndex for KeywordIndex {

    fn get(&self, key: &str) -> Option<Doc> {

        if let Some(doc) = self.buffer.get(key) {

            return Some(doc.clone());

        }

        self.lookup(self.fields.key, key)

    }

    fn by_content(&self, content: &str) -> Option<Doc> {

        if let Some(doc) = self.buffer.values().find(|doc| doc.content == content) {

            return Some(doc.clone());

        }

        self.lookup(self.fields.content, content)

    }

    fn put(&mut self, doc: Doc) {

        self.remove(&doc.key);

        self.buffer.insert(doc.key.clone(), doc);

    }

    fn remove(&mut self, key: &str) -> bool {

        let buffered = self.buffer.remove(key).is_some();

        let key_field = self.fields.key;

        self.remove_where(key_field, key) || buffered

    }

    fn remove_provisional(&mut self, target: &str) -> bool {

        let prefix = format!("{target}\n");

        let before = self.buffer.len();

        self.buffer

            .retain(|key, doc| !(doc.provisional && key.starts_with(&prefix)));

        let buffered = self.buffer.len() != before;

        let target_field = self.fields.target;

        self.remove_where(target_field, target) || buffered

    }

    fn buffered(&self) -> usize {

        self.buffer.len()

    }

    fn search(

        &mut self,

        query: &str,

        rooms: Option<&[String]>,

        before: Option<(u64, u64)>,

        limit: usize,

    ) -> Result<(Vec<Hit>, Vec<String>), Error> {

        let (text_query, terms) = build_query(self.fields.text, query)?;

        self.flush_buffer()?;

        let query: Box<dyn Query> = match rooms {

            None => text_query,

            Some(rooms) => Box::new(BooleanQuery::new(vec![

                (Occur::Must, text_query),

                (

                    Occur::Must,

                    Box::new(TermSetQuery::new(

                        rooms

                            .iter()

                            .map(|room| Term::from_field_text(self.fields.room, room)),

                    )),

                ),

            ])),

        };

        let mut found = Vec::new();

        for segment in &self.segments {

            let collector = Newest {

                alive: segment.alive.clone(),

                before,

                limit,

            };

            let newest = segment

                .searcher

                .search(&*query, &collector)

                .map_err(|error| tantivy_error("searching messages", error))?;

            for (ts, tie, doc) in newest {

                found.push((ts, tie, segment.id, doc));

            }

        }

        found.sort_by(|a, b| (b.0, b.1).cmp(&(a.0, a.1)));

        found.truncate(limit);

        let mut hits = Vec::new();

        for (ts, tie, id, doc) in found {

            let segment = self

                .segments

                .iter()

                .find(|segment| segment.id == id)

                .expect("the segment was just searched");

            let document: TantivyDocument = segment

                .searcher

                .doc(doc)

                .map_err(|error| tantivy_error("reading a search hit", error))?;

            let Some(stored) = self.stored(&document) else {

                continue;

            };

            hits.push(Hit {

                key: stored.key,

                room: stored.room,

                sender: stored.sender,

                content: stored.content,

                ts,

                tie,

            });

        }

        Ok((hits, terms))

    }

    fn maintain(&mut self) -> Result<(), Error> {

        self.flush_buffer()?;

        self.compact()

    }

    fn changes(&mut self) -> Result<Changes, Error> {

        self.maintain()?;

        let mut changes = Changes::default();

        for (id, persisted, del_gen) in &self.removed {

            if *persisted {

                changes.deletes.push(segment_key(*id));

            }

            if *del_gen > 0 {

                changes.deletes.push(deletes_key(*id, *del_gen));

            }

            changes.removed.push(*id);

        }

        let mut records = Vec::with_capacity(self.segments.len());

        for segment in &self.segments {

            if !segment.persisted {

                changes

                    .puts

                    .push((segment_key(segment.id), Self::segment_files(segment)?));

                changes.files.push(segment.id);

            }

            let mut del_gen = segment.del_gen;

            if segment.deletes_dirty && segment.live() < segment.max_doc as usize {

                del_gen = segment.del_gen + 1;

                let mut bytes = Vec::new();

                write_alive_bitset(&segment.alive, &mut bytes)

                    .map_err(|error| tantivy_error("saving search deletes", error))?;

                changes.puts.push((deletes_key(segment.id, del_gen), bytes));

                if segment.del_gen > 0 {

                    changes

                        .deletes

                        .push(deletes_key(segment.id, segment.del_gen));

                }

                changes.dels.push((segment.id, del_gen));

            }

            records.push(SegmentRecord {

                id: segment.id,

                del_gen,

            });

        }

        changes.manifest = manifest(records.into_iter(), self.next_id);

        Ok(changes)

    }

    fn committed(&mut self, changes: &Changes) {

        self.removed

            .retain(|(id, ..)| !changes.removed.contains(id));

        for segment in &mut self.segments {

            if changes.files.contains(&segment.id) {

                segment.persisted = true;

            }

            if let Some((_, generation)) = changes.dels.iter().find(|(id, _)| *id == segment.id) {

                segment.del_gen = *generation;

                segment.deletes_dirty = false;

            }

        }

    }

    fn detach(&mut self) {

        self.persistent = false;

        self.removed.clear();

    }

}

/// the `limit` newest live docs older than `before`, by timestamp and then key hash

struct Newest {

    alive: Arc<BitSet>,

    before: Option<(u64, u64)>,

    limit: usize,

}

struct NewestSegment {

    alive: Arc<BitSet>,

    before: Option<(u64, u64)>,

    limit: usize,

    ts: tantivy::columnar::Column<u64>,

    tie: tantivy::columnar::Column<u64>,

    segment: u32,

    top: std::collections::BinaryHeap<std::cmp::Reverse<(u64, u64, DocAddress)>>,

}

impl Collector for Newest {

    type Fruit = Vec<(u64, u64, DocAddress)>;

    type Child = NewestSegment;

    fn for_segment(&self, segment: u32, reader: &SegmentReader) -> tantivy::Result<NewestSegment> {

        Ok(NewestSegment {

            alive: self.alive.clone(),

            before: self.before,

            limit: self.limit,

            ts: reader.fast_fields().u64("ts")?,