yazi/yazi-watcher/src/remote/remote.rs

use std::time::{Duration, SystemTime};

use hashbrown::HashMap;
use notify::Result;
use tokio::{pin, sync::mpsc::UnboundedReceiver};
use tokio_stream::{StreamExt, wrappers::UnboundedReceiverStream};
use yazi_fs::{FilesOp, file::File};
use yazi_shared::url::{UrlBuf, UrlLike};
use yazi_vfs::VfsFile;

use crate::{MgrProxy, Reporter, WATCHER, Watchee};

pub(crate) struct Remote;

impl Remote {
	pub(crate) fn serve(rx: UnboundedReceiver<(UrlBuf, bool)>, _reporter: Reporter) -> Self {
		tokio::spawn(Self::changed(rx));

		Self
	}

	pub(crate) fn watch(&mut self, _watchee: &mut Watchee) -> Result<()> { Ok(()) }

	pub(crate) fn unwatch(&mut self, _watchee: &Watchee) -> Result<()> { Ok(()) }

	async fn changed(rx: UnboundedReceiver<(UrlBuf, bool)>) {
		let rx = UnboundedReceiverStream::new(rx).chunks_timeout(1000, Duration::from_millis(250));
		pin!(rx);

		while let Some(chunk) = rx.next().await {
			let mut urls = HashMap::with_capacity(chunk.len());
			for (u, upload) in chunk {
				urls.entry(u).and_modify(|b| *b |= upload).or_insert(upload);
			}

			let _permit = WATCHER.acquire().await.unwrap();

			let mut ops = Vec::with_capacity(urls.len());
			let mut ups = Vec::with_capacity(urls.len());

			for (u, upload) in urls {
				let Some((parent, key)) = u.pair2() else { continue };
				let Ok(mut file) = File::new(&u).await else {
					ops.push(FilesOp::Deleting(parent.into(), [key.into()].into()));
					continue;
				};

				let is_file = file.is_file();
				file.cha.ctime = Some(SystemTime::now());

				ops.push(FilesOp::Upserting(parent.into(), [(key.into(), file)].into()));
				if upload && is_file {
					ups.push(u);
				}
			}

			FilesOp::mutate(ops);
			MgrProxy::upload(ups);
		}
	}
}