diff --git a/src/server.js b/src/server.js
index 3e35d2c..33ce6cb 100644
--- a/src/server.js
+++ b/src/server.js
@@ -23,7 +23,7 @@ const handleIntent = async (streamKey, intent) => {
 	console.log("TODO: Handle Intents", intent);
 }
 
-const processor = async (group, consumer, streamKey) => {
+async function processor(group, consumer, streamKey) {
 	await redis.client.connect().catch(console.error);
 
 	while (true) {
@@ -38,16 +38,18 @@ const processor = async (group, consumer, streamKey) => {
 
 		for (const stream of res) {
 			for (const msg of stream.messages) {
+				try {
 				console.log(`[${consumer}] Processing:`, msg);
 
 				await handleIntent("commands", msg);
 
 				await redis.client.xAck(streamKey, group, msg.id);
+				} catch (err) {
+					console.error("Failed to process message:", err);
+				}
 			}
 		}
 	}
-
-	await redis.client.disconnect();
 }
 
 const main = async () => {
@@ -55,12 +57,13 @@ const main = async () => {
 	const doForce = config.FORCE_DB_SYNC === "true" && config.NODE_ENV !== "production";
 	await sequelize.authenticate();
 	await sequelize.sync({force: doForce});
+
 	if (doForce) console.info("Database dropped and resynced (force=true).");
 
 	await redis.ensureGroup("transcripts", 'nlp');
 	await redis.ensureGroup("commands", 'nlp');
 
-	processor('nlp', `${config.DEVICE_ID}`, "transcripts");
+	processor('nlp', `${config.DEVICE_ID}`, "transcripts").catch(console.error);
 }
 
 main().catch(console.error);