<commit_before>package com.github.rnewson.couchdb.lucene;

import static java.util.concurrent.TimeUnit.SECONDS;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.SocketException;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;

import net.sf.json.JSONObject;

import org.apache.commons.lang.StringUtils;
import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.HttpStatus;
import org.apache.http.client.HttpClient;
import org.apache.http.client.HttpResponseException;
import org.apache.http.client.ResponseHandler;
import org.apache.http.client.methods.HttpUriRequest;
import org.apache.log4j.Logger;
import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.document.Document;
import org.apache.lucene.index.IndexReader;
import org.apache.lucene.index.IndexWriter;
import org.apache.lucene.index.Term;
import org.mozilla.javascript.ClassShutter;
import org.mozilla.javascript.Context;
import org.mozilla.javascript.RhinoException;

import com.github.rnewson.couchdb.lucene.Lucene.ReaderCallback;
import com.github.rnewson.couchdb.lucene.Lucene.WriterCallback;
import com.github.rnewson.couchdb.lucene.couchdb.Couch;
import com.github.rnewson.couchdb.lucene.couchdb.Database;
import com.github.rnewson.couchdb.lucene.util.Analyzers;
import com.github.rnewson.couchdb.lucene.util.Constants;
import com.github.rnewson.couchdb.lucene.util.IndexPath;

public final class ViewIndexer implements Runnable {

    private final class RestrictiveClassShutter implements ClassShutter {
        public boolean visibleToScripts(final String fullClassName) {
            return false;
        }
    }

    private class ViewChangesHandler implements ResponseHandler<Void> {

        private final Analyzer analyzer;
        private final DocumentConverter converter;
        private final JSONObject defaults;
        private final String digest;
        private HttpUriRequest request;
        private final long latchThreshold;
        private boolean pendingCommit;
        private long pendingSince;
        private long since;

        public ViewChangesHandler(final UUID uuid, final JSONObject view, final long latchThreshold) throws IOException {
            this.latchThreshold = latchThreshold;
            this.defaults = view.has("defaults") ? view.getJSONObject("defaults") : defaults();
            this.analyzer = Analyzers.getAnalyzer(view.optString("analyzer", "standard"));
            final String function = extractFunction(view);
            this.converter = new DocumentConverter(context, null, function);
            lucene.createWriter(path, uuid, function);
            this.digest = Lucene.digest(function);
            lucene.withReader(path, false, new ReaderCallback() {
                public void callback(final IndexReader reader) throws IOException {
                    final Map<String, String> commit = reader.getCommitUserData();
                    if (commit != null && commit.containsKey("last_seq")) {
                        since = Long.parseLong(commit.get("last_seq"));
                    }
                }

                public void onMissing() throws IOException {
                    since = 0;
                }
            });
            releaseCatch();
        }

        public Void handleResponse(final HttpResponse response) {
            try {
                final HttpEntity entity = response.getEntity();
                final BufferedReader reader = new BufferedReader(new InputStreamReader(entity.getContent(), "UTF-8"));
                String line;
                loop: while ((line = reader.readLine()) != null) {
                    commitDocuments();

                    // Heartbeat.
                    if (line.length() == 0) {
                        logger.trace("heartbeat");
                        continue loop;
                    }

                    final JSONObject json = JSONObject.fromObject(line);

                    if (json.has("error")) {
                        logger.warn("Indexing stopping due to error: " + json);
                        break loop;
                    }

                    if (json.has("last_seq")) {
                        logger.warn("End of changes detected.");
                        break loop;
                    }

                    final String id = json.getString("id");
                    JSONObject doc;
                    if (json.has("doc")) {
                        doc = json.getJSONObject("doc");
                    } else {
                        // include_docs=true doesn't work prior to 0.11.
                        try {
                            doc = database.getDocument(id);
                        } catch (final HttpResponseException e) {
                            switch (e.getStatusCode()) {
                            case HttpStatus.SC_NOT_FOUND:
                                doc = JSONObject.fromObject("{_deleted:true}");
                                break;
                            default:
                                logger.warn("Failed to fetch " + id);
                                break loop;
                            }
                        }
                    }

                    if (id.equals("_design/" + path.getDesignDocumentName())) {
                        if (doc.optBoolean("_deleted")) {
                            logger.info("Design document for this view was deleted.");
                            break loop;
                        }
                        final JSONObject view = extractView(doc);
                        if (view == null) {
                            logger.info("View was deleted.");
                            break loop;
                        }
                        final String fun = extractFunction(view);
                        if (fun == null) {
                            logger.warn("View has no index function.");
                            break loop;
                        }
                        final String newDigest = Lucene.digest(fun);
                        if (!digest.equals(newDigest)) {
                            logger.info("Digest of function changed.");
                            break loop;
                        }
                    } else if (id.startsWith("_design")) {
                        // Ignore other design document changes.
                        continue loop;
                    } else if (doc.optBoolean("_deleted")) {
                        logger.trace(id + " deleted.");
                        lucene.withWriter(path, new WriterCallback() {

                            public boolean callback(final IndexWriter writer) throws IOException {
                                writer.deleteDocuments(new Term("_id", id));
                                return true;
                            }

                            public void onMissing() throws IOException {
                                // Ignore.
                            }
                        });
                        setPendingCommit(true);
                    } else {
                        logger.trace(id + " updated.");
                        final Document[] docs;
                        try {
                            docs = converter.convert(doc, defaults, database);
                        } catch (final RhinoException e) {
                            logger.warn(id + " caused " + e.getMessage());
                            continue loop;
                        }

                        lucene.withWriter(path, new WriterCallback() {

                            public boolean callback(final IndexWriter writer) throws IOException {
                                writer.deleteDocuments(new Term("_id", id));
                                for (final Document doc : docs) {
                                    if (logger.isTraceEnabled()) {
                                        logger.trace(id + " -> " + doc);
                                    }
                                    writer.addDocument(doc, analyzer);
                                }
                                return true;
                            }

                            public void onMissing() throws IOException {
                                // Ignore.
                            }
                        });
                        setPendingCommit(true);
                    }
                    since = json.getLong("seq");
                    releaseCatch();<commit_msg>don't emit misleading 'committed checkpoint' statement if no documents were added at all.<commit_after>package com.github.rnewson.couchdb.lucene;

import static java.util.concurrent.TimeUnit.SECONDS;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.SocketException;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;

import net.sf.json.JSONObject;

import org.apache.commons.lang.StringUtils;
import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.HttpStatus;
import org.apache.http.client.HttpClient;
import org.apache.http.client.HttpResponseException;
import org.apache.http.client.ResponseHandler;
import org.apache.http.client.methods.HttpUriRequest;
import org.apache.log4j.Logger;
import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.document.Document;
import org.apache.lucene.index.IndexReader;
import org.apache.lucene.index.IndexWriter;
import org.apache.lucene.index.Term;
import org.mozilla.javascript.ClassShutter;
import org.mozilla.javascript.Context;
import org.mozilla.javascript.RhinoException;

import com.github.rnewson.couchdb.lucene.Lucene.ReaderCallback;
import com.github.rnewson.couchdb.lucene.Lucene.WriterCallback;
import com.github.rnewson.couchdb.lucene.couchdb.Couch;
import com.github.rnewson.couchdb.lucene.couchdb.Database;
import com.github.rnewson.couchdb.lucene.util.Analyzers;
import com.github.rnewson.couchdb.lucene.util.Constants;
import com.github.rnewson.couchdb.lucene.util.IndexPath;

public final class ViewIndexer implements Runnable {

    private final class RestrictiveClassShutter implements ClassShutter {
        public boolean visibleToScripts(final String fullClassName) {
            return false;
        }
    }

    private class ViewChangesHandler implements ResponseHandler<Void> {

        private final Analyzer analyzer;
        private final DocumentConverter converter;
        private final JSONObject defaults;
        private final String digest;
        private HttpUriRequest request;
        private final long latchThreshold;
        private boolean pendingCommit;
        private long pendingSince;
        private long since;

        public ViewChangesHandler(final UUID uuid, final JSONObject view, final long latchThreshold) throws IOException {
            this.latchThreshold = latchThreshold;
            this.defaults = view.has("defaults") ? view.getJSONObject("defaults") : defaults();
            this.analyzer = Analyzers.getAnalyzer(view.optString("analyzer", "standard"));
            final String function = extractFunction(view);
            this.converter = new DocumentConverter(context, null, function);
            lucene.createWriter(path, uuid, function);
            this.digest = Lucene.digest(function);
            lucene.withReader(path, false, new ReaderCallback() {
                public void callback(final IndexReader reader) throws IOException {
                    final Map<String, String> commit = reader.getCommitUserData();
                    if (commit != null && commit.containsKey("last_seq")) {
                        since = Long.parseLong(commit.get("last_seq"));
                    }
                }

                public void onMissing() throws IOException {
                    since = 0;
                }
            });
            releaseCatch();
        }

        public Void handleResponse(final HttpResponse response) {
            try {
                final HttpEntity entity = response.getEntity();
                final BufferedReader reader = new BufferedReader(new InputStreamReader(entity.getContent(), "UTF-8"));
                String line;
                loop: while ((line = reader.readLine()) != null) {
                    commitDocuments();

                    // Heartbeat.
                    if (line.length() == 0) {
                        logger.trace("heartbeat");
                        continue loop;
                    }

                    final JSONObject json = JSONObject.fromObject(line);

                    if (json.has("error")) {
                        logger.warn("Indexing stopping due to error: " + json);
                        break loop;
                    }

                    if (json.has("last_seq")) {
                        logger.warn("End of changes detected.");
                        break loop;
                    }

                    final String id = json.getString("id");
                    JSONObject doc;
                    if (json.has("doc")) {
                        doc = json.getJSONObject("doc");
                    } else {
                        // include_docs=true doesn't work prior to 0.11.
                        try {
                            doc = database.getDocument(id);
                        } catch (final HttpResponseException e) {
                            switch (e.getStatusCode()) {
                            case HttpStatus.SC_NOT_FOUND:
                                doc = JSONObject.fromObject("{_deleted:true}");
                                break;
                            default:
                                logger.warn("Failed to fetch " + id);
                                break loop;
                            }
                        }
                    }

                    if (id.equals("_design/" + path.getDesignDocumentName())) {
                        if (doc.optBoolean("_deleted")) {
                            logger.info("Design document for this view was deleted.");
                            break loop;
                        }
                        final JSONObject view = extractView(doc);
                        if (view == null) {
                            logger.info("View was deleted.");
                            break loop;
                        }
                        final String fun = extractFunction(view);
                        if (fun == null) {
                            logger.warn("View has no index function.");
                            break loop;
                        }
                        final String newDigest = Lucene.digest(fun);
                        if (!digest.equals(newDigest)) {
                            logger.info("Digest of function changed.");
                            break loop;
                        }
                    } else if (id.startsWith("_design")) {
                        // Ignore other design document changes.
                        continue loop;
                    } else if (doc.optBoolean("_deleted")) {
                        logger.trace(id + " deleted.");
                        lucene.withWriter(path, new WriterCallback() {

                            public boolean callback(final IndexWriter writer) throws IOException {
                                writer.deleteDocuments(new Term("_id", id));
                                return true;
                            }

                            public void onMissing() throws IOException {
                                // Ignore.
                            }
                        });
                        setPendingCommit(true);
                    } else {
                        logger.trace(id + " updated.");
                        final Document[] docs;
                        try {
                            docs = converter.convert(doc, defaults, database);
                        } catch (final RhinoException e) {
                            logger.warn(id + " caused " + e.getMessage());
                            continue loop;
                        }

                        lucene.withWriter(path, new WriterCallback() {

                            public boolean callback(final IndexWriter writer) throws IOException {
                                writer.deleteDocuments(new Term("_id", id));
                                for (final Document doc : docs) {
                                    if (logger.isTraceEnabled()) {
                                        logger.trace(id + " -> " + doc);
                                    }
                                    writer.addDocument(doc, analyzer);
                                }
                                return true;
                            }

                            public void onMissing() throws IOException {
                                // Ignore.
                            }
                        });
                        if (docs.length > 0) {
                            setPendingCommit(true);
                        }
                    }
                    since = json.getLong("seq");
                    releaseCatch();
{'fraction_non_alphanumeric': 0.08477974451509424, 'fraction_numerical': 0.0008592541673827118, 'mean_word_length': 1.2871741124066554, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 10, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2142680', 'n_tokens_mistral': 4045, 'n_tokens_neox': 3781, 'n_words': 872}