Explore Search Docs Sign in Get started

Docs Pricing Support Changelog

coline/onset

O

colineonset

Public

Onset: version control where nobody waits and nothing gets lost. The on command, its core and onset-server.

Sign in to fork

CodeIssuesChange requestsActionsDiscussionsCommitsBranchesReleasesSearchContributorsCompare

Code Issues Change requests Actions Discussions Commits Branches Releases Search Contributors Compare

onset/server/src/api.rs

Radin feat(vcs): live on vcs.coline.dev; fewer round trips, small objects in the database, batched git view

23bb7718dfd5·…History

277 lines · 12 KBtext/plain; charset=utf-8

1	//! HTTP: routes requests to the state, one SQLite connection per thread.
2
3	use std::io::Read;
4	use std

::

path

::

Path

;

5	use std::sync::Arc;

7	use anyhow::Result;

8	use serde::Serialize;

9	use serde_json::json;

10	use tiny_http::{Header, Method, Request, Response, Server};

11	use vcs_cli::host::LandRequest;

12	use vcs_cli::wire::{self, Hold, MineFile, PutChange, Rule, SaveRecord, ShareRequest};

13	use vcs_core::Id;

15	use crate::state::{Access, Problem, Res, State, problem};

17	/// Biggest request body accepted (a bundle of a big change).

18	const MAX_BODY: u64 = 512 << 20;

20	pub fn serve(data: &Path, listen: &str, threads: usize) -> Result<()> {

21	// Create the schema once before the workers start.

22	State::open(data)?;

23	let server = Arc::new(Server::http(listen).map_err(\|e\| anyhow::anyhow!("can't listen on {listen}: {e}"))?);

24	eprintln!("vcs-server on http://{listen}, data in {}", data.display());

25	let mut handles = Vec::new();

26	for _ in 0..threads.max(1) {

27	let server = server.clone();

28	let data = data.to_path_buf();

29	handles.push(std::thread::spawn(move \|\| {

30	let state = State::open(&data).expect("open state");

31	for request in server.incoming_requests() {

32	handle(&state, request);

}

}));

}

36	for h in handles {

37	let _ = h.join();

}

Ok(())

}

42	enum Body {

43	Json(serde_json::Value),

44	Bytes(Vec<u8>),

}

47	fn handle(state: &State, mut request: Request) {

let started = std::time::Instant::now();

49	let method = request.method().clone();

50	let url = request.url().to_string();

51	let result = route(state, &mut request);

42	enum Body {

53	Ok((status, body)) => (status, body),

54	Err(p) => (p.status, Body::Json(json!({ "error": { "code": p.code, "message": p.message, "holder": p.holder } }))),

};

56	let response = match body {

57	Body::Json(v) => Response::from_data(serde_json::to_vec(&v).unwrap_or_default())

58	.with_status_code(status)

59	.with_header(Header::from_bytes("content-type", "application/json").unwrap()),

60	Body::Bytes(b) => Response::from_data(b).with_status_code(status).with_header(Header::from_bytes("content-type", "application/octet-stream").unwrap()),

};

62	if std::env::var_os("VCS_SERVER_LOG").is_some() {

63	eprintln!("{method} {url} → {status} in {:?}", started.elapsed());

}

65	let _ = request.respond(response);

}

68	fn ok<T: Serialize>(value: T) -> Res<(u16, Body)> {

69	Ok((200, Body::Json(serde_json::to_value(value).map_err(\|e\| problem(500, "failed", e.to_string()))?)))

}

72	fn read_body(request: &mut Request) -> Res<Vec<u8>> {

73	let mut body = Vec::new();

74	request.as_reader().take(MAX_BODY + 1).read_to_end(&mut body).map_err(\|e\| problem(400, "bad_usage", e.to_string()))?;

75	if body.len() as u64 > MAX_BODY {

76	return Err(problem(413, "too_big", "request too big"));

}

Ok(body)

}

81	fn json_body<T: serde::de::DeserializeOwned>(request: &mut Request) -> Res<T> {

82	let body = read_body(request)?;

83	serde_json::from_slice(&body).map_err(\|e\| problem(400, "bad_usage", format!("bad request body: {e}")))

}

86	fn query(url: &str, key: &str) -> Option<String> {

87	url.split_once('?')?.1.split('&').find_map(\|kv\| kv.strip_prefix(&format!("{key}=")).map(str::to_string))

}

90	fn route(state: &State, request: &mut Request) -> Res<(u16, Body)> {

91	let url = request.url().to_string();

92	let path = url.split('?').next().unwrap_or("");

93	let method = request.method().clone();

94	if path == "/.well-known/vcs.json" {

95	return ok(json!({ "api": "/v1" }));

}

97	let Some(rest) = path.strip_prefix("/v1/") else { return Err(problem(404, "not_found", "no such endpoint")) };

98	let parts: Vec<&str> = rest.split('/').filter(\|p\| !p.is_empty()).collect();

100	let token = request

101	.headers()

94	if path == "/.well-known/vcs.json" {

103	.find(\|h\| h.field.equiv("authorization"))

.and_then(|h| h.value.as_str().strip_prefix("Bearer ").map(str::to_string));

105	let user = match token {

106	Some(t) => Some(state.user_for_token(&t)?.ok_or_else(\|\| problem(401, "not_logged_in", "this token isn't valid here"))?),

107	None => None,

};

110	match (&method, parts.as_slice()) {

111	(Method::Get, ["me"]) => {

112	let me = user.ok_or_else(\|\| problem(401, "not_logged_in", "sign in first"))?;

ok(me)

}

115	(Method::Post, ["projects"]) => {

116	let me = user.ok_or_else(\|\| problem(401, "not_logged_in", "sign in first"))?;

107	None => None,

118	struct New {

119	owner: String,

120	name: String,

121	#[serde(default)]

122	public: bool,

}

124	let new: New = json_body(request)?;

125	if !state.may_create_under(&new.owner, &me.handle)? {

126	return Err(problem(403, "not_allowed", format!("{} belongs to someone else", new.owner)));

}

128	let key = format!("{}/{}", new.owner, new.name);

129	if state.project_exists(&key)?.is_some() {

130	return Err(problem(409, "exists", format!("{key} already exists")));

}

132	state.add_project(&key, &me.handle, new.public)?;

133	let access = state.access(&key, Some(me))?;

134	ok(view(state, &key, &access)?)

}

136	(_, ["projects", owner, name, tail @ ..]) => {

137	let key = format!("{owner}/{name}");

138	let access = state.access(&key, user)?;

139	project_route(state, request, &method, &key, &access, tail, &url)

}

141	_ => Err(problem(404, "not_found", "no such endpoint")),

}

}

145	fn view(state: &State, key: &str, access: &Access) -> Res<wire::ProjectView> {

146	let (owner, name) = key.split_once('/').unwrap();

147	Ok(wire::ProjectView { owner: owner.into(), name: name.into(), public: state.public(key)?, main: state.main_for(key, access)?, can: access.can.clone(), git: None, git_stopped: None })

}

150	fn need_user(access: &Access) -> Res<()> {

151	if access.user.is_none() {

152	return Err(problem(401, "not_logged_in", "sign in first"));

}

Ok(())

}

157	fn project_route(state: &State, request: &mut Request, method: &Method, key: &str, access: &Access, tail: &[&str], url: &str) -> Res<(u16, Body)> {

158	match (method, tail) {

159	(Method::Get, []) => ok(view(state, key, access)?),

160	(Method::Get, ["versions"]) => {

161	let since: u64 = query(url, "since").and_then(\|s\| s.parse().ok()).unwrap_or(0);

162	let limit: i64 = query(url, "limit").and_then(\|s\| s.parse().ok()).unwrap_or(5000).min(5000);

163	let versions = state.versions_since(key, since, limit)?;

164	ok(json!({ "versions": state.versions_for(key, access, versions)? }))

}

166	(Method::Post, ["objects", "missing"]) => {

167	need_user(access)?;

168	let ids = wire::decode_ids(&read_body(request)?);

169	Ok((200, Body::Bytes(wire::encode_ids(&state.missing(&ids)))))

}

171	(Method::Post, ["objects"]) => {

172	need_user(access)?;

173	let body = read_body(request)?;

174	let stored = state.store.ingest(&body).map_err(\|e\| problem(400, "bad_usage", format!("{e:#}")))?;

175	ok(json!({ "stored": stored }))

}

177	(Method::Get, ["trees", tree]) => {

178	let tree = Id::parse(tree).ok_or_else(\|\| problem(400, "bad_usage", "not a tree id"))?;

179	if !state.may_read_tree(key, access, &tree)? {

180	return Err(problem(404, "not_found", "no such tree"));

}

182	let have = query(url, "have").and_then(\|h\| Id::parse(&h)).filter(\|h\| state.may_read_tree(key, access, h).unwrap_or(false));

183	Ok((200, Body::Bytes(state.tree_bundle(&tree, have.as_ref())?)))

}

185	(Method::Get, ["changes"]) => ok(json!({ "changes": state.changes(key, access)? })),

186	(Method::Get, ["changes", prefix]) => {

187	let (change, saves) = state.find_change(key, access, prefix)?;

188	ok(json!({ "change": change, "saves": saves }))

}

190	(Method::Put, ["changes", id]) => {

191	let put: PutChange = json_body(request)?;

192	if put.change.id != *id {

193	return Err(problem(400, "bad_usage", "change id mismatch"));

}

195	let change = state.put_change(key, access, put.change, &put.workspace, &put.label)?;

196	if let Some(save) = &put.save {

197	state.add_save(key, access, id, &SaveRecord { workspace: put.workspace.clone(), ..save.clone() })?;

191	let put: PutChange = json_body(request)?;

199	ok(json!({ "change": change }))

}

201	(Method::Post, ["changes", id, "hold"]) => {

202	let hold: Hold = json_body(request)?;

203	ok(json!({ "previous": state.hold(key, access, id, &hold.workspace, &hold.label)? }))

}

205	(Method::Post, ["changes", id, "saves"]) => {

206	let save: SaveRecord = json_body(request)?;

207	state.add_save(key, access, id, &save)?;

208	ok(json!({}))

}

210	(Method::Post, ["changes", id, "access"]) => {

211	let share: ShareRequest = json_body(request)?;

212	ok(json!({ "shared": state.share(key, access, id, &share.who, share.edit)? }))

}

214	(Method::Post, ["import"]) => {

215	#[derive(serde::Deserialize)]

216	struct Import {

217	versions: Vec<vcs_core::version::Version>,

}

219	let import: Import = json_body(request)?;

220	let main = state.import(key, access, import.versions)?;

221	ok(json!({ "main": main }))

}

223	// This server has no git view: git IDs from an import aren't needed.

224	(Method::Post, ["git", "ids"]) => {

225	if !access.can.admin {

226	return Err(problem(403, "not_allowed", "only the project's owners can do that"));

}

228	let _ = read_body(request)?;

229	ok(json!({ "added": 0 }))

}

231	(Method::Post, ["git", "resume"]) => Err(problem(400, "bad_usage", "this server has no git views")),

232	(Method::Post, ["land"]) => {

233	let land: LandRequest = json_body(request)?;

234	// The change's record and newest save come along with the landing.

235	if let Some(put) = &land.record {

236	if put.change.id != land.change {

237	return Err(problem(400, "bad_usage", "change id mismatch"));

}

239	state.put_change(key, access, put.change.clone(), &put.workspace, &put.label)?;

240	if let Some(save) = &put.save {

241	state.add_save(key, access, &land.change, &SaveRecord { workspace: put.workspace.clone(), ..save.clone() })?;

}

}

244	ok(state.land(key, access, &land)?)

}

246	(Method::Get, ["rules"]) => ok(json!({ "rules": state.rules(key)? })),

247	(Method::Put, ["rules"]) => {

248	let rule: Rule = json_body(request)?;

249	ok(json!({ "rules": state.set_rule(key, access, &rule)? }))

}

251	(Method::Get, ["mine"]) => {

252	need_user(access)?;

253	ok(json!({ "files": state.mine(key, &access.user.as_ref().unwrap().handle)? }))

}

255	(Method::Put, ["mine"]) => {

256	need_user(access)?;

248	let rule: Rule = json_body(request)?;

258	struct Files {

259	files: Vec<MineFile>,

}

261	let files: Files = json_body(request)?;

262	state.set_mine(key, &access.user.as_ref().unwrap().handle, &files.files)?;

263	ok(json!({}))

}

265	(Method::Get, ["mine", "bundle"]) => {

266	need_user(access)?;

267	let files = state.mine(key, &access.user.as_ref().unwrap().handle)?;

268	let mut ids = Vec::new();

269	for f in files {

270	ids.extend(state.store.file_parts(&f.file)?);

}

272	Ok((200, Body::Bytes(state.store.bundle(&ids)?)))

}

274	_ => Err(Problem { status: 404, code: "not_found", message: "no such endpoint".into(), holder: None }),

}

}