
/*
 * Copyright (C) 2015 Archie L. Cobbs. All rights reserved.
 */

package org.jsimpledb.jsck.cmd;

import java.io.PrintWriter;
import java.util.Collections;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

import org.jsimpledb.Session;
import org.jsimpledb.SessionMode;
import org.jsimpledb.cli.CliSession;
import org.jsimpledb.cli.cmd.AbstractCommand;
import org.jsimpledb.core.FieldTypeRegistry;
import org.jsimpledb.jsck.Jsck;
import org.jsimpledb.jsck.JsckConfig;
import org.jsimpledb.jsck.JsckLogger;
import org.jsimpledb.kv.KVTransaction;
import org.jsimpledb.parse.expr.Node;
import org.jsimpledb.schema.SchemaModel;
import org.jsimpledb.util.ParseContext;

public class JsckCommand extends AbstractCommand {

    public JsckCommand() {
        super("jsck"
         + " -repair:repair"
         + " -verbose:verbose"
         + " -weak:weak"
         + " -limit:limit:int"
         + " -gc:garbageCollectSchemas"
         + " -force-schemas:schema-map:expr"
         + " -force-format-version:format-version:int"
         + " -registry:registry:expr");
    }

    @Override
    public String getHelpSummary() {
        return "Check a key/value database for inconsistencies and optionally repair them";
    }

    @Override
    public String getHelpDetail() {
        return "Options:\n"
          + "   -repair\n"
          + "       In addition to detecting issues, attempt to repair them.\n"
          + "   -limit\n"
          + "       Stop after encountering `limit' issues.\n"
          + "   -gc\n"
          + "       Garbage collect unused schema versions at the end of inspection.\n"
          + "   -registry\n"
          + "       Specify a custom field type registry. If this flag is not given, in JSimpleDB and Core API modes,\n"
          + "       the configured registry will be used; in key/value database CLI mode, a default instances is used.\n"
          + "       The parameter must be a Java expression returning a FieldTypeRegistry.\n"
          + "   -force-schemas\n"
          + "       Forcibly override schema versions. The parameter must be a Java expression returning a\n"
          + "       Map<Integer, SchemaModel>. WARNING: only use this if you know what you are doing.\n"
          + "       This flag is ignored without `-repair'.\n"
          + "   -force-format-version\n"
          + "       Forcibly override format version. WARNING: only use this if you know what you are doing.\n"
          + "       This flag is ignored without `-repair'.\n"
          + "   -verbose\n"
          + "       Increase logging verbosity to show a high level of detail.\n"
          + "   -weak\n"
          + "       For certain key/value stores, use weaker consistency to reduce the chance of conflicts.\n"
          + "       This flag is ignored if used with `-repair'.\n";
    }

    @Override
    public EnumSet<SessionMode> getSessionModes() {
        return EnumSet.allOf(SessionMode.class);
    }

    @Override
    public CliSession.Action getAction(CliSession session, ParseContext ctx, boolean complete, Map<String, Object> params) {

        // Setup config (partially)
        final JsckConfig config = new JsckConfig();
        final boolean verbose = params.containsKey("verbose");
        config.setGarbageCollectSchemas(params.containsKey("gc"));
        config.setRepair(params.containsKey("repair"));
        final boolean weak = params.containsKey("weak");
        final Integer formatVersion = (Integer)params.get("force-format-version");
        if (formatVersion != null)
            config.setForceFormatVersion(formatVersion);
        final Integer limit = (Integer)params.get("limit");
        if (limit != null)
            config.setMaxIssues(limit);

        // Done
        return new JsckAction(config, (Node)params.get("registry"), (Node)params.get("force-schemas"), verbose, weak);
    }

    private class JsckAction implements CliSession.Action, Session.TransactionalAction, Session.HasTransactionOptions {

        private final JsckConfig config;
        private final Node registryNode;
        private final Node schemasNode;
        private final boolean verbose;
        private final boolean weak;

        JsckAction(JsckConfig config, Node registryNode, Node schemasNode, boolean verbose, boolean weak) {
            this.config = config;
            this.registryNode = registryNode;
            this.schemasNode = schemasNode;
            this.verbose = verbose;
            this.weak = weak;
        }

        @Override
        public void run(CliSession session) throws Exception {

            // Evaluate registry, if any
            if (this.registryNode != null) {
                this.config.setFieldTypeRegistry(
                  JsckCommand.this.getExprParam(session, this.registryNode, "registry", FieldTypeRegistry.class));
            } else if (session.getDatabase() != null)
                config.setFieldTypeRegistry(session.getDatabase().getFieldTypeRegistry());

            // Evaluate forced schemas, if any
            if (this.schemasNode != null) {
                config.setForceSchemaVersions(JsckCommand.this.getExprParam(session, this.schemasNode, "force-schemas", obj -> {
                    if (!(obj instanceof Map))
                        throw new IllegalArgumentException("must be a Map<Integer, SchemaModel>");
                    final Map<?, ?> map = (Map<?, ?>)obj;
                    final HashMap<Integer, SchemaModel> versionMap = new HashMap<>(map.size());
                    for (Map.Entry<?, ?> entry : map.entrySet()) {
                        if (!(entry.getKey() instanceof Integer))
                            throw new IllegalArgumentException("must be a Map<Integer, SchemaModel>; found key " + entry.getKey());
                        if ((entry.getValue() != null && !(entry.getValue() instanceof SchemaModel))) {
                            throw new IllegalArgumentException("must be a Map<Integer, SchemaModel>; found value "
                              + entry.getValue());
                        }
                        versionMap.put((Integer)entry.getKey(), (SchemaModel)entry.getValue());
                    }
                    return versionMap;
                }));
            }

            // Configure logger to log to console
            final PrintWriter writer = session.getWriter();
            this.config.setJsckLogger(new JsckLogger() {
                @Override
                public boolean isDetailEnabled() {
                    return verbose;
                }
                @Override
                public void info(String message) {
                    writer.println("jsck: " + message);
                }
                @Override
                public void detail(String message) {
                    if (verbose)
                        writer.println("jsck: " + message);
                }
            });

            // Do scan
            final KVTransaction kvt = session.getKVTransaction();
            final Jsck jsck = new Jsck(this.config);
            final AtomicInteger count = new AtomicInteger();
            final long numHandled = jsck.check(kvt, issue -> {
                writer.println(String.format("[%05d] %s%s", count.incrementAndGet(),
                  issue, this.config.isRepair() ? " [FIXED]" : ""));
            });
            writer.println("jsck: " + (this.config.isRepair() ? "repaired" : "found") + " " + numHandled + " issue(s)");
        }

        // Use EVENTUAL_COMMITTED consistency for Raft key/value stores to avoid retries
        @Override
        public Map<String, ?> getTransactionOptions() {
            return this.weak && !this.config.isRepair() ? Collections.singletonMap("consistency", "EVENTUAL") : null;
        }
    }
}


@@ -971,36 +971,17 @@
  %22 -gc:g
-arbageCollectSchemas
+c
 %22%0A      
@@ -1194,18 +1194,16 @@
  %22Check 
-a 
 key/valu
@@ -3522,22 +3522,16 @@
 ms.get(%22
-force-
 format-v

Fix bugs in jsck CLI command parameter parsing.